148 lines
5.3 KiB
Python
148 lines
5.3 KiB
Python
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()
|