120 lines
3.6 KiB
Python
120 lines
3.6 KiB
Python
"""unit test: add_products_to_database stamps newListing and copies it onto alerts
|
|
|
|
run from the repo root:
|
|
uv run pytest tests/test_new_listing_stamp.py
|
|
|
|
offline — drives add_products_to_database against fake collections that record
|
|
the UpdateOne ops instead of touching Mongo. The recorded ops are inspected via
|
|
UpdateOne's private _doc/_filter, the same fields its own __eq__ compares.
|
|
|
|
The stamp is the database layer's job: only it knows whether the doc already
|
|
exists, so monitors never set the flag. The alert carries its own copy because
|
|
the product doc is rewritten every cycle (its flag flips False one cycle after
|
|
discovery) while a discovery alert may outlive that in a backed-up queue.
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
from monitor_utils.database import add_products_to_database, build_alert
|
|
from monitor_utils.models import PricingInfo, Product
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
NOW = datetime(2026, 8, 10, tzinfo=timezone.utc)
|
|
|
|
|
|
class FakeCursor:
|
|
def __init__(self, docs):
|
|
self._docs = list(docs)
|
|
|
|
def __aiter__(self):
|
|
return self
|
|
|
|
async def __anext__(self):
|
|
if not self._docs:
|
|
raise StopAsyncIteration
|
|
return self._docs.pop(0)
|
|
|
|
|
|
class FakeCollection:
|
|
"""Records bulk_write ops; doubles as its own .database for alerts lookup."""
|
|
|
|
def __init__(self, existing_docs=(), name="products"):
|
|
self.name = name
|
|
self._existing = list(existing_docs)
|
|
self.bulk_writes = []
|
|
self.database = self
|
|
self.alerts = None
|
|
|
|
def __getitem__(self, name):
|
|
if self.alerts is None:
|
|
self.alerts = FakeCollection(name=name)
|
|
return self.alerts
|
|
|
|
def find(self, _filter, _projection):
|
|
return FakeCursor(self._existing)
|
|
|
|
async def bulk_write(self, ops, ordered=True):
|
|
self.bulk_writes.append(list(ops))
|
|
|
|
|
|
def _product(product_id="p1", current=95.0, original=100.0) -> Product:
|
|
return Product(
|
|
product_id=product_id,
|
|
title="Widget",
|
|
url="http://x/p1",
|
|
pricing=PricingInfo(current_price=current, original_price=original),
|
|
)
|
|
|
|
|
|
def run(collection, products):
|
|
asyncio.run(
|
|
add_products_to_database(products, collection, "tools", "bestbuy")
|
|
)
|
|
|
|
|
|
def product_update(collection):
|
|
return collection.bulk_writes[0][0]._doc
|
|
|
|
|
|
def alert_doc(collection):
|
|
return collection.alerts.bulk_writes[0][0]._doc["$setOnInsert"]
|
|
|
|
|
|
def test_insert_stamps_new_listing_true_on_doc_and_alert():
|
|
collection = FakeCollection()
|
|
run(collection, [_product()])
|
|
|
|
assert product_update(collection)["$set"]["newListing"] is True
|
|
assert alert_doc(collection)["newListing"] is True
|
|
|
|
|
|
def test_existing_doc_with_new_price_stamps_false():
|
|
existing = {"_id": "p1", "currentPrice": 100.0, "pastPrices": []}
|
|
collection = FakeCollection([existing])
|
|
run(collection, [_product(current=95.0)])
|
|
|
|
assert product_update(collection)["$set"]["newListing"] is False
|
|
# a never-seen price still alerts, but not as a new listing
|
|
assert alert_doc(collection)["newListing"] is False
|
|
|
|
|
|
def test_existing_doc_with_known_price_emits_no_alert():
|
|
existing = {
|
|
"_id": "p1",
|
|
"currentPrice": 95.0,
|
|
"pastPrices": [{"price": 95.0, "date": NOW}],
|
|
}
|
|
collection = FakeCollection([existing])
|
|
run(collection, [_product(current=95.0)])
|
|
|
|
assert product_update(collection)["$set"]["newListing"] is False
|
|
assert collection.alerts is None
|
|
|
|
|
|
def test_build_alert_defaults_missing_flag_to_false():
|
|
"""Direct callers passing an unstamped document must not KeyError."""
|
|
|
|
alert = build_alert("bestbuy", {"_id": "p1"}, NOW)
|
|
assert alert["newListing"] is False
|