test: cover propertyfinder matching rules

.gitea/workflows/ci.yml

@@ -32,4 +32,6 @@ jobs:
     needs: hygiene
     steps:
       - uses: actions/checkout@v4
+      - run: python3 -m pip install -r requirements.txt
       - run: python3 -m compileall app
+      - run: python3 -m unittest discover -s tests

app/worker.py

@@ -10,6 +10,8 @@ import json
 import sys
 from typing import Any
 
+from sqlalchemy import text
+
 from app.db import SessionLocal, init_db
 from app.models import Project
 from app.services.monitor import (
@@ -131,7 +133,17 @@ def cmd_suggest(payload: dict[str, Any]) -> None:
         db.close()
 
 
+def cmd_health(_: dict[str, Any]) -> None:
+    db = SessionLocal()
+    try:
+        db.execute(text("SELECT 1"))
+        _write({"status": "ok"})
+    finally:
+        db.close()
+
+
 COMMANDS = {
+    "health": cmd_health,
     "add-listing": cmd_add_listing,
     "add-listings": cmd_add_listings,
     "check-project": cmd_check_project,

go.mod

@@ -1,8 +1,11 @@
 module monitoring-pf
 
-go 1.25.0
+go 1.25.7
 
-require modernc.org/sqlite v1.50.1
+require (
+	gitea.estateliga.work/admin/portal-common v0.3.0
+	modernc.org/sqlite v1.50.1
+)
 
 require (
 	github.com/dustin/go-humanize v1.0.1 // indirect

go.sum

@@ -1,3 +1,5 @@
+gitea.estateliga.work/admin/portal-common v0.3.0 h1:xpr9UeLXk5pCcNXcTVGZzJZr0Ni7An7DV0OkuYv9qVM=
+gitea.estateliga.work/admin/portal-common v0.3.0/go.mod h1:C860q6g38KVMsv+mKv6k1Vm7smVRCycl+N6r63TElnk=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
 github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs=

internal/pf/http.go

@@ -1,12 +1,16 @@
 package pf
 
 import (
+	"context"
 	"database/sql"
 	"encoding/json"
 	"errors"
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
+
+	commonmw "gitea.estateliga.work/admin/portal-common/middleware"
 )
 
 type Server struct {
@@ -21,17 +25,32 @@ type bulkPayload struct {
 	URLs []string `json:"urls"`
 }
 
+type componentProbe struct {
+	Name      string `json:"name"`
+	Status    string `json:"status"`
+	LatencyMs int64  `json:"latency_ms"`
+	Error     string `json:"error,omitempty"`
+}
+
 func (s Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	path := s.apiPath(r.URL.Path)
 	switch {
 	case path == "/healthz":
 		writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
+	case path == "/health/detail":
+		s.healthDetail(w, r)
 	case path == "/":
 		writeJSON(w, http.StatusOK, map[string]string{"service": "monitoring-pf", "ui": "portal", "api": "go"})
 	case !strings.HasPrefix(path, "/api/v1"):
 		writeError(w, http.StatusNotFound, "not found")
-	case !s.checkInternalAuth(w, r):
-		return
+	default:
+		commonmw.InternalAuth(s.App.Cfg.InternalAPIKey)(http.HandlerFunc(s.serveAPI)).ServeHTTP(w, r)
+	}
+}
+
+func (s Server) serveAPI(w http.ResponseWriter, r *http.Request) {
+	path := s.apiPath(r.URL.Path)
+	switch {
 	case path == "/api/v1/access/me" && r.Method == http.MethodGet:
 		s.accessMe(w, r)
 	case path == "/api/v1/summary" && r.Method == http.MethodGet:
@@ -53,18 +72,6 @@ func (s Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func (s Server) checkInternalAuth(w http.ResponseWriter, r *http.Request) bool {
-	want := strings.TrimSpace(s.App.Cfg.InternalAPIKey)
-	if want == "" {
-		return true
-	}
-	if r.Header.Get("X-Internal-Key") != want {
-		writeError(w, http.StatusUnauthorized, "unauthorized")
-		return false
-	}
-	return true
-}
-
 func (s Server) apiPath(path string) string {
 	base := s.App.Cfg.PublicBasePath
 	if base != "" && path == base {
@@ -76,6 +83,104 @@ func (s Server) apiPath(path string) string {
 	return path
 }
 
+func (s Server) healthDetail(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
+	defer cancel()
+
+	components := []componentProbe{
+		s.probeDatabase(ctx),
+		s.probeWorker(ctx),
+		s.probeTelegram(ctx),
+		s.probeScheduler(ctx),
+		s.probeProjectIntegrity(ctx),
+	}
+	writeJSON(w, http.StatusOK, map[string]any{"components": components})
+}
+
+func (s Server) probeDatabase(ctx context.Context) componentProbe {
+	start := time.Now()
+	if err := s.App.DB.PingContext(ctx); err != nil {
+		return componentProbe{Name: "sqlite", Status: "down", LatencyMs: time.Since(start).Milliseconds(), Error: err.Error()}
+	}
+	return componentProbe{Name: "sqlite", Status: "ok", LatencyMs: time.Since(start).Milliseconds()}
+}
+
+func (s Server) probeWorker(ctx context.Context) componentProbe {
+	start := time.Now()
+	if err := s.App.Worker.Health(ctx); err != nil {
+		return componentProbe{Name: "python_worker", Status: "down", LatencyMs: time.Since(start).Milliseconds(), Error: err.Error()}
+	}
+	return componentProbe{Name: "python_worker", Status: "ok", LatencyMs: time.Since(start).Milliseconds()}
+}
+
+func (s Server) probeTelegram(ctx context.Context) componentProbe {
+	start := time.Now()
+	if !s.App.TG.Enabled() {
+		return componentProbe{Name: "telegram_bot", Status: "down", LatencyMs: time.Since(start).Milliseconds(), Error: "TG_BOT_TOKEN is not configured"}
+	}
+	if _, err := s.App.TG.BotUsername(ctx); err != nil {
+		return componentProbe{Name: "telegram_bot", Status: "down", LatencyMs: time.Since(start).Milliseconds(), Error: err.Error()}
+	}
+	return componentProbe{Name: "telegram_bot", Status: "ok", LatencyMs: time.Since(start).Milliseconds()}
+}
+
+func (s Server) probeScheduler(ctx context.Context) componentProbe {
+	start := time.Now()
+	staleAfter := max(1, s.App.Cfg.ScrapeIntervalHours*2) * 3600
+	staleModifier := "-" + strconv.Itoa(staleAfter) + " seconds"
+	var total, neverChecked, stale int64
+	err := s.App.DB.QueryRowContext(ctx, `
+		SELECT
+			COUNT(*) AS total,
+			COALESCE(SUM(CASE WHEN last_checked_at IS NULL THEN 1 ELSE 0 END), 0) AS never_checked,
+			COALESCE(SUM(CASE WHEN last_checked_at IS NOT NULL AND datetime(last_checked_at) < datetime('now', ?) THEN 1 ELSE 0 END), 0) AS stale
+		FROM projects`,
+		staleModifier,
+	).Scan(&total, &neverChecked, &stale)
+	if err != nil {
+		return componentProbe{Name: "scheduler", Status: "down", LatencyMs: time.Since(start).Milliseconds(), Error: err.Error()}
+	}
+	if total > 0 && (neverChecked > 0 || stale > 0) {
+		return componentProbe{
+			Name:      "scheduler",
+			Status:    "down",
+			LatencyMs: time.Since(start).Milliseconds(),
+			Error: "projects=" + strconv.FormatInt(total, 10) +
+				" never_checked=" + strconv.FormatInt(neverChecked, 10) +
+				" stale=" + strconv.FormatInt(stale, 10) +
+				" stale_after_sec=" + strconv.Itoa(staleAfter),
+		}
+	}
+	return componentProbe{Name: "scheduler", Status: "ok", LatencyMs: time.Since(start).Milliseconds()}
+}
+
+func (s Server) probeProjectIntegrity(ctx context.Context) componentProbe {
+	start := time.Now()
+	var projects, missingOwner, autoWithoutPermit, searchURLs int64
+	err := s.App.DB.QueryRowContext(ctx, `
+		SELECT
+			(SELECT COUNT(*) FROM projects),
+			(SELECT COUNT(*) FROM projects p LEFT JOIN employees e ON e.id = p.owner_id WHERE e.id IS NULL),
+			(SELECT COUNT(*) FROM competitor_listings WHERE auto_discovered = 1 AND (permit_number IS NULL OR trim(permit_number) = '')),
+			(SELECT COUNT(*) FROM competitor_listings WHERE url LIKE '%/search%' OR url LIKE '%?%' AND external_id = '')
+	`).Scan(&projects, &missingOwner, &autoWithoutPermit, &searchURLs)
+	if err != nil {
+		return componentProbe{Name: "project_integrity", Status: "down", LatencyMs: time.Since(start).Milliseconds(), Error: err.Error()}
+	}
+	if missingOwner > 0 || autoWithoutPermit > 0 || searchURLs > 0 {
+		return componentProbe{
+			Name:      "project_integrity",
+			Status:    "down",
+			LatencyMs: time.Since(start).Milliseconds(),
+			Error: "projects=" + strconv.FormatInt(projects, 10) +
+				" missing_owner=" + strconv.FormatInt(missingOwner, 10) +
+				" auto_without_permit=" + strconv.FormatInt(autoWithoutPermit, 10) +
+				" suspicious_search_urls=" + strconv.FormatInt(searchURLs, 10),
+		}
+	}
+	return componentProbe{Name: "project_integrity", Status: "ok", LatencyMs: time.Since(start).Milliseconds()}
+}
+
 func (s Server) accessMe(w http.ResponseWriter, r *http.Request) {
 	portalID := portalUserID(r)
 	emp, err := s.App.CurrentEmployee(r.Context(), portalID, false)
@@ -510,11 +615,11 @@ func portalUserID(r *http.Request) string {
 }
 
 func isAdmin(r *http.Request) bool {
-	return r.Header.Get("X-User-Is-Admin") == "1"
+	return commonmw.HeaderBool(r, "X-User-Is-Admin")
 }
 
 func canViewTeam(r *http.Request) bool {
-	return isAdmin(r) || r.Header.Get("X-User-Is-Department-Head") == "1"
+	return isAdmin(r) || commonmw.HeaderBool(r, "X-User-Is-Department-Head")
 }
 
 func canManagePortalUser(r *http.Request, targetPortalID string) bool {
@@ -534,19 +639,7 @@ func canManagePortalUser(r *http.Request, targetPortalID string) bool {
 }
 
 func subordinatePortalIDs(r *http.Request) []string {
-	raw := strings.TrimSpace(r.Header.Get("X-User-Subordinates"))
-	if raw == "" {
-		return []string{}
-	}
-	parts := strings.Split(raw, ",")
-	out := make([]string, 0, len(parts))
-	for _, part := range parts {
-		id := strings.TrimSpace(part)
-		if id != "" {
-			out = append(out, id)
-		}
-	}
-	return out
+	return commonmw.HeaderCSV(r, "X-User-Subordinates")
 }
 
 func ownerPortalIDFromQuery(r *http.Request) *string {

internal/pf/store_test.go

@@ -0,0 +1,67 @@
+package pf
+
+import (
+	"strings"
+	"testing"
+)
+
+func strPtr(v string) *string {
+	return &v
+}
+
+func int64Ptr(v int64) *int64 {
+	return &v
+}
+
+func float64Ptr(v float64) *float64 {
+	return &v
+}
+
+func validProjectPayload() ProjectPayload {
+	return ProjectPayload{
+		Title:     "Full Park View",
+		DealType:  "sale",
+		OurPrice:  float64Ptr(2500000),
+		DLDPermit: strPtr("7140504127"),
+		Building:  strPtr("Harbour Gate Tower 2"),
+		Bedrooms:  int64Ptr(2),
+		SizeSqft:  float64Ptr(1081),
+		OurURL: strPtr(
+			"https://www.propertyfinder.ae/en/plp/buy/apartment-for-sale-dubai-dubai-creek-harbour-the-lagoons-harbour-gate-harbour-gate-tower-2-86176216.html",
+		),
+	}
+}
+
+func TestValidateProjectRequiredAcceptsConcretePropertyFinderListingURL(t *testing.T) {
+	payload := validProjectPayload()
+
+	if err := validateProjectRequired(payload); err != nil {
+		t.Fatalf("validateProjectRequired() returned unexpected error: %v", err)
+	}
+}
+
+func TestValidateProjectRequiredRejectsSearchPageAsOurURL(t *testing.T) {
+	payload := validProjectPayload()
+	payload.OurURL = strPtr("https://www.propertyfinder.ae/en/search?c=1&l=12345")
+
+	err := validateProjectRequired(payload)
+	if err == nil {
+		t.Fatal("validateProjectRequired() accepted a search page as our_url")
+	}
+	if !strings.Contains(err.Error(), "concrete PropertyFinder listing URL") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestValidateProjectRequiredRejectsListingLikeURLWithoutID(t *testing.T) {
+	payload := validProjectPayload()
+	payload.OurURL = strPtr("https://www.propertyfinder.ae/en/plp/buy/apartment-for-sale-dubai-dubai-creek-harbour.html")
+
+	err := validateProjectRequired(payload)
+	if err == nil {
+		t.Fatal("validateProjectRequired() accepted a listing-like URL without listing id")
+	}
+	if !strings.Contains(err.Error(), "concrete PropertyFinder listing URL") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}

internal/pf/worker.go

@@ -34,6 +34,10 @@ type CheckResult struct {
 	Changes int `json:"changes"`
 }
 
+type HealthResult struct {
+	Status string `json:"status"`
+}
+
 type Suggestion struct {
 	Source       string   `json:"source"`
 	ExternalID   string   `json:"external_id"`
@@ -104,6 +108,17 @@ func (w *Worker) Suggest(ctx context.Context, projectID int64) (*SuggestionsResp
 	return &out, nil
 }
 
+func (w *Worker) Health(ctx context.Context) error {
+	var out HealthResult
+	if err := w.call(ctx, "health", map[string]any{}, &out); err != nil {
+		return err
+	}
+	if out.Status != "ok" {
+		return fmt.Errorf("worker status=%s", out.Status)
+	}
+	return nil
+}
+
 func (w *Worker) call(ctx context.Context, command string, payload any, out any) error {
 	ctx, cancel := context.WithTimeout(ctx, 15*time.Minute)
 	defer cancel()

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()