test: cover propertyfinder matching rules

tests/test_monitoring_rules.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.db import Base
+from app.models import CompetitorListing, DealType, Employee, ListingStatus, Project, Source
+from app.scrapers.base import ScrapedListing
+from app.scrapers.propertyfinder import PropertyFinderScraper
+from app.services import monitor
+
+
+PF_OWN_URL = (
+    "https://www.propertyfinder.ae/en/plp/buy/apartment-for-sale-dubai-dubai-creek-harbour-"
+    "the-lagoons-harbour-gate-harbour-gate-tower-2-86176216.html"
+)
+PF_COMPETITOR_URL = (
+    "https://www.propertyfinder.ae/en/plp/buy/apartment-for-sale-dubai-dubai-creek-harbour-"
+    "the-lagoons-harbour-gate-harbour-gate-tower-2-86170000.html"
+)
+
+
+def _listing(external_id: str, permit: str | None, url: str = PF_COMPETITOR_URL) -> ScrapedListing:
+    return ScrapedListing(
+        source="propertyfinder",
+        external_id=external_id,
+        url=url,
+        title=f"Listing {external_id}",
+        price=2_500_000,
+        currency="AED",
+        permit_number=permit,
+        agent_name="Agent",
+        agency_name="Agency",
+        is_active=True,
+    )
+
+
+class MonitoringRulesTest(unittest.TestCase):
+    def setUp(self) -> None:
+        engine = create_engine("sqlite:///:memory:", future=True)
+        Base.metadata.create_all(engine)
+        self.Session = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
+        self.db = self.Session()
+
+        owner = Employee(name="Agent", portal_user_id="agent-1", tg_chat_id="100")
+        self.db.add(owner)
+        self.db.flush()
+        self.project = Project(
+            title="Full Park View",
+            deal_type=DealType.SALE,
+            our_price=2_500_000,
+            dld_permit="7140504127",
+            building="Harbour Gate Tower 2",
+            bedrooms=2,
+            size_sqft=1081,
+            our_url=PF_OWN_URL,
+            owner_id=owner.id,
+        )
+        self.db.add(self.project)
+        self.db.commit()
+
+    def tearDown(self) -> None:
+        self.db.close()
+
+    def test_propertyfinder_rejects_search_pages(self) -> None:
+        scraper = PropertyFinderScraper()
+
+        self.assertFalse(scraper.is_listing_url("https://www.propertyfinder.ae/en/search?c=1&l=12345"))
+        self.assertIsNone(scraper.fetch_listing("https://www.propertyfinder.ae/en/search?c=1&l=12345"))
+
+    @patch.object(monitor.PF, "get_permit", side_effect=["7140504127"])
+    @patch.object(
+        monitor.PF,
+        "search_similar",
+        return_value=[
+            _listing("86176216", None, url=PF_OWN_URL),
+            _listing("86170000", None, url=PF_COMPETITOR_URL),
+        ],
+    )
+    def test_suggest_similar_excludes_own_listing(self, _search, _permit) -> None:
+        suggestions = monitor.suggest_similar(self.project, our_permit="7140504127")
+
+        self.assertEqual(["86170000"], [item.external_id for item in suggestions["propertyfinder"]])
+
+    @patch.object(
+        monitor,
+        "suggest_similar",
+        return_value={
+            "propertyfinder": [
+                _listing("86170000", "7140504127"),
+                _listing("86170001", "DIFFERENT"),
+            ],
+            "bayut": [],
+        },
+    )
+    def test_sync_permit_competitors_adds_only_exact_permit_matches(self, _suggest) -> None:
+        changes, suggestions, permit = monitor.sync_permit_competitors(self.db, self.project)
+
+        listings = self.db.query(CompetitorListing).order_by(CompetitorListing.external_id).all()
+        self.assertEqual("7140504127", permit)
+        self.assertEqual(1, len(listings))
+        self.assertEqual("86170000", listings[0].external_id)
+        self.assertTrue(listings[0].auto_discovered)
+        self.assertEqual(["86170001"], [item.external_id for item in suggestions["propertyfinder"]])
+        self.assertEqual(1, len(changes))
+
+    @patch.object(monitor, "suggest_similar", return_value={"propertyfinder": [], "bayut": []})
+    def test_auto_permit_listing_is_removed_only_after_three_misses(self, _suggest) -> None:
+        listing = CompetitorListing(
+            project_id=self.project.id,
+            source=Source.PROPERTYFINDER,
+            external_id="86170000",
+            url=PF_COMPETITOR_URL,
+            title="Competitor",
+            permit_number="7140504127",
+            auto_discovered=True,
+            permit_missing_checks=0,
+            current_price=2_500_000,
+            currency="AED",
+            status=ListingStatus.ACTIVE,
+        )
+        self.db.add(listing)
+        self.db.commit()
+
+        changes, _, _ = monitor.sync_permit_competitors(self.db, self.project)
+        self.db.flush()
+        self.assertEqual([], changes)
+        self.assertEqual(1, self.db.query(CompetitorListing).count())
+        self.assertEqual(1, listing.permit_missing_checks)
+
+        changes, _, _ = monitor.sync_permit_competitors(self.db, self.project)
+        self.db.flush()
+        self.assertEqual([], changes)
+        self.assertEqual(1, self.db.query(CompetitorListing).count())
+        self.assertEqual(2, listing.permit_missing_checks)
+
+        changes, _, _ = monitor.sync_permit_competitors(self.db, self.project)
+        self.db.flush()
+        self.assertEqual(1, len(changes))
+        self.assertEqual(0, self.db.query(CompetitorListing).count())
+
+
+if __name__ == "__main__":
+    unittest.main()