MarketingTool/tests/test_app.py

import json
import os
import tempfile
import threading
import time
import urllib.error
import unittest
from http.client import HTTPConnection
from unittest.mock import patch

from api import server
from api import browser_investigator


class BrowserOnlyAppTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.NamedTemporaryFile(delete=False)
        self.tmp.close()
        server.DB_PATH = self.tmp.name
        self.scan_patcher = patch("api.server.scan_website", return_value={"website_status": "healthy", "website_http_status": 200, "website_final_url": "", "website_latency_ms": 50, "website_https": 1, "website_title": "", "website_technology": "", "website_has_viewport": 1, "website_tls_expires_at": "", "website_checked_at": server.now(), "website_error": "", "website_emails": [], "email_evidence": []})
        self.scan_patcher.start()
        self.http = server.ThreadingHTTPServer(("127.0.0.1", 0), server.Handler)
        self.thread = threading.Thread(target=self.http.serve_forever, daemon=True)
        self.thread.start()
        self.conn = HTTPConnection("127.0.0.1", self.http.server_port)

    def tearDown(self):
        self.scan_patcher.stop()
        self.http.shutdown(); self.http.server_close(); self.thread.join(1)
        try: os.unlink(self.tmp.name)
        except FileNotFoundError: pass

    def request(self, method, path, payload=None):
        body = json.dumps(payload).encode() if payload is not None else None
        self.conn.request(method, path, body, {"Content-Type": "application/json"} if body else {})
        response = self.conn.getresponse()
        return response.status, json.loads(response.read() or b"{}")

    def wait_for_run(self, run_id):
        for _ in range(300):
            status, body = self.request("GET", f"/api/runs/{run_id}")
            self.assertEqual(status, 200)
            if body["status"] not in {"queued", "running"}:
                return body
            time.sleep(0.02)
        self.fail("run did not finish")

    def test_search_redirect_links_are_unwrapped(self):
        wrapped = "https://duckduckgo.com/l/?uddg=https%3A%2F%2Fwww.example.co.za%2Fservices"
        self.assertEqual(server.normalize_result_href(wrapped), "https://www.example.co.za/services")
        self.assertEqual(server.normalize_result_href("//example.co.za/about"), "https://example.co.za/about")

    @patch.dict(os.environ, {"DISCOVERY_BACKEND": "browser"})
    @patch("api.server.google_search", return_value=[{"name": "Acme Plumbing", "website": "https://www.acme.co.za", "domain": "acme.co.za"}, {"name": "Other", "website": "https://other.co.za", "domain": "other.co.za"}])
    def test_discovery_filters_catalogued_root_domains(self, search):
        status, first = self.request("POST", "/api/discover", {"category": "plumbers", "keywords": "plumbing", "location": "Cape Town", "max_results": 5})
        self.assertEqual(status, 202)
        first = self.wait_for_run(first["id"])
        self.assertEqual(first["added"], 2)
        saved = self.request("GET", "/api/prospects")[1]["items"]
        self.assertTrue(all(item["search_category"] == "plumbers" for item in saved))
        status, second = self.request("POST", "/api/discover", {"category": "plumbers", "keywords": "plumbing", "location": "Cape Town", "max_results": 5})
        self.assertEqual(status, 202)
        second = self.wait_for_run(second["id"])
        self.assertEqual(second["found"], 2); self.assertEqual(second["added"], 0)
        self.assertEqual(self.request("GET", "/api/prospects")[1]["total"], 2)

    @patch.dict(os.environ, {"DISCOVERY_BACKEND": "browser"})
    def test_map_without_website_gets_stable_catalogue_key(self):
        mapped = server.map_gmaps_result({"title": "A Local Plumber", "place_id": "abc123", "address": "Cape Town", "phone": "+27 10", "review_rating": 4.5, "review_count": 12}, "plumbers Cape Town")
        self.assertEqual(mapped["domain"], "maps:abc123")
        self.assertEqual(mapped["website_status"], "no_detected_website")
        self.assertGreaterEqual(mapped["opportunity_score"], 70)

    @patch.dict(os.environ, {"DISCOVERY_BACKEND": "browser"})
    def test_review_updates_only_selected_prospect(self):
        with patch("api.server.google_search", return_value=[{"name": "Acme", "website": "https://acme.co.za", "domain": "acme.co.za"}]):
            _, run = self.request("POST", "/api/discover", {"category": "plumbers", "keywords": "plumbing", "location": "South Africa"})
            self.wait_for_run(run["id"])
        status, item = self.request("PATCH", "/api/prospects/1/review", {"status": "approved", "notes": "Looks relevant"})
        self.assertEqual(status, 200); self.assertEqual(item["review_status"], "approved")

    def test_standard_maps_worker_uses_jobs_api_without_key(self):
        csv_body = b"title,address,website,phone,latitude,longitude,review_rating,review_count,link\nAcme Plumbing,Cape Town,https://acme.co.za,+2710, -33.9,18.4,4.5,12,https://maps.google.com/?cid=1\n"
        with patch.object(server, "GMAPS_URL", "http://gmaps:8080"), \
             patch.object(server, "GMAPS_API_MODE", "standard"), \
             patch.object(server, "GMAPS_API_KEY", ""), \
             patch("api.server.http_json", side_effect=[{"id": "job-1"}, {"ID": "job-1", "Status": "pending"}, {"ID": "job-1", "Status": "working"}, {"ID": "job-1", "Status": "ok"}]) as http_json, \
             patch("api.server.http_bytes", return_value=csv_body) as http_bytes:
            job_id = server.submit_gmaps_job("plumbers Cape Town")
            results = server.poll_gmaps_job(job_id)
        self.assertEqual(job_id, "job-1")
        self.assertEqual(results[0]["title"], "Acme Plumbing")
        first_call = http_json.call_args_list[0]
        self.assertEqual(first_call.args[0:2], ("POST", "http://gmaps:8080/api/v1/jobs"))
        self.assertNotIn("X-API-Key", first_call.args[3])
        self.assertTrue(first_call.args[2]["email"])
        self.assertEqual(http_bytes.call_args.args[1], "http://gmaps:8080/api/v1/jobs/job-1/download")

    def test_nous_provider_exposes_models_and_tests_connection(self):
        free_models = [{"id": "meituan/longcat-2.0:free", "label": "Meituan · LongCat 2.0", "free": True}]
        with patch.object(server, "nous_models", return_value=free_models):
            status, body = self.request("GET", "/api/ai/models?provider=nous")
            self.assertEqual(status, 200)
            self.assertEqual(body["models"], free_models)
            status, body = self.request("POST", "/api/settings", {"ai": {"AI_PROVIDER": "nous", "AI_MODEL": "meituan/longcat-2.0:free", "AI_API_KEY": "nous-test-key"}})
            self.assertEqual(status, 200)
            self.assertEqual(body["ai"]["provider"], "nous")
            status, body = self.request("GET", "/api/settings")
            self.assertEqual(status, 200)
            self.assertTrue(body["ai"]["configured"])
            status, body = self.request("GET", "/api/config")
            self.assertEqual(status, 200)
            self.assertEqual(body["ai_provider"], "nous")
            self.assertNotIn("nous-test-key", json.dumps(body))

        with patch("api.server.http_json", return_value={"choices": [{"message": {"content": "OK"}}]}) as http_json:
            status, body = self.request("POST", "/api/ai/test", {})
            self.assertEqual(status, 200)
            self.assertTrue(body["ok"])
            self.assertEqual(http_json.call_args.args[0:2], ("POST", "https://inference-api.nousresearch.com/v1/chat/completions"))

    def test_provider_errors_are_not_mislabelled_as_maps_errors(self):
        class CloudflareResponse:
            def read(self):
                return b'{"error_code":1010,"detail":"blocked"}'

            def close(self):
                return None

            code = 403

        with patch("api.server.urllib.request.urlopen", side_effect=urllib.error.HTTPError("https://inference-api.nousresearch.com/v1/chat/completions", 403, "Forbidden", {}, CloudflareResponse())):
            with self.assertRaisesRegex(RuntimeError, r"^ai_cloudflare_1010:"):
                server.http_json("POST", "https://inference-api.nousresearch.com/v1/chat/completions", {}, error_prefix="ai")

    def test_provider_requests_send_explicit_user_agent(self):
        response = type("Response", (), {"__enter__": lambda self: self, "__exit__": lambda self, *args: None, "read": lambda self: b"{}"})()
        with patch("api.server.urllib.request.urlopen", return_value=response) as open_url:
            server.http_json("GET", "https://example.test/v1/models", error_prefix="ai")
        self.assertEqual(open_url.call_args.args[0].headers.get("User-agent"), server.OUTBOUND_USER_AGENT)

    def test_ai_assessment_is_saved_for_a_prospect(self):
        conn = server.db()
        conn.execute("INSERT INTO prospects(name,domain,category,website_status,opportunity_score,discovered_at) VALUES(?,?,?,?,?,?)", ("Acme Plumbing", "acme.co.za", "plumber", "no_detected_website", 82, server.now()))
        conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", ("AI_PROVIDER", "nous", server.now()))
        conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", ("AI_MODEL", "meituan/longcat-2.0:free", server.now()))
        conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", ("AI_API_KEY", "test-key", server.now()))
        conn.commit(); conn.close()
        response = {"choices": [{"message": {"content": '{"probability":84,"recommendation":"contact","reasoning":"Active business with a clear website opportunity."}'}}]}
        with patch("api.server.http_json", return_value=response):
            status, body = self.request("POST", "/api/prospects/1/assess", {})
        self.assertEqual(status, 200)
        self.assertEqual(body["ai_probability"], 84)
        self.assertEqual(body["ai_recommendation"], "contact")
        saved = server.db().execute("SELECT ai_probability,ai_recommendation,ai_reasoning FROM prospects WHERE id=1").fetchone()
        self.assertEqual(tuple(saved), (84.0, "contact", "Active business with a clear website opportunity."))

    def test_browser_investigator_normalises_and_filters_candidates(self):
        raw = '{"candidates":[{"url":"https://www.acme.co.za/about","confidence":88,"evidence":["name match"]},{"url":"https://facebook.com/acme","confidence":99},{"domain":"other.co.za","confidence":"42","evidence":"not a list"}]}'
        candidates = browser_investigator._parse_agent_result(raw)
        self.assertEqual([item["domain"] for item in candidates], ["acme.co.za", "other.co.za"])
        self.assertEqual(candidates[0]["confidence"], 88)
        self.assertEqual(candidates[1]["evidence"], [])

    def test_browser_website_candidates_can_be_found_and_accepted(self):
        conn = server.db()
        conn.execute("INSERT INTO prospects(name,domain,category,address,discovered_at) VALUES(?,?,?,?,?)", ("Acme Plumbing", "maps:abc", "plumber", "Cape Town", server.now()))
        conn.commit(); conn.close()
        found = [{"url": "https://acmeplumbing.co.za", "domain": "acmeplumbing.co.za", "confidence": 91, "evidence": ["name match", "location match"]}]
        with patch("api.server.investigate_websites", return_value=found):
            status, body = self.request("POST", "/api/prospects/1/website-candidates", {})
        self.assertEqual(status, 200)
        self.assertEqual(body["count"], 1)
        candidate_id = body["items"][0]["id"]
        status, body = self.request("GET", "/api/prospects/1/website-candidates")
        self.assertEqual(status, 200)
        self.assertEqual(body["items"][0]["domain"], "acmeplumbing.co.za")
        status, body = self.request("PATCH", f"/api/prospects/1/website-candidates/{candidate_id}", {"status": "accepted"})
        self.assertEqual(status, 200)
        self.assertEqual(body["candidate"]["status"], "accepted")
        self.assertEqual(body["prospect"]["domain"], "acmeplumbing.co.za")
        self.assertEqual(body["prospect"]["website"], "https://acmeplumbing.co.za")

    def test_email_draft_returns_structured_review_notes(self):
        conn = server.db()
        conn.execute("INSERT INTO prospects(name,domain,search_category,category,address,maps_url,website_status,emails,opportunity_score,score_reasons,discovered_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)", ("Acme Plumbing", "maps:abc", "plumbers", "Plumber", "Cape Town", "https://maps.example/acme", "no_detected_website", "acme@gmail.com", 88, '["No detected website","Public email found"]', server.now()))
        for key, value in (("AI_PROVIDER", "custom"), ("AI_BASE_URL", "https://ai.example/v1"), ("AI_MODEL", "test-model"), ("AI_API_KEY", "test-key")):
            conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", (key, value, server.now()))
        conn.commit(); conn.close()
        content = json.dumps({"subject": "A website idea for Acme Plumbing", "message": "Good day Acme Plumbing,\n\nI noticed your public listing does not show a dedicated website. We can help with a professional online presence. Would you like me to send a few details?\n\nKind regards", "rationale": "Uses the missing-website signal and the supplied offer.", "recommended_offer": "Gold Plus hosting and a free one-page website", "unknowns": ["Confirm the preferred contact person"], "evidence_used": ["prospect_record", "public_maps_listing", "public_email_listing"], "follow_up": "Confirm the offer and recipient before sending."})
        with patch("api.server.http_json", return_value={"choices": [{"message": {"content": content}}]}):
            status, body = self.request("POST", "/api/prospects/1/message", {"offer": "Gold Plus hosting at R119/month with a free one-page website."})
        self.assertEqual(status, 200)
        self.assertEqual(body["unknowns"], ["Confirm the preferred contact person"])
        self.assertEqual(body["evidence_used"], ["prospect_record", "public_maps_listing", "public_email_listing"])
        saved = server.db().execute("SELECT ai_unknowns,ai_evidence_ids,ai_follow_up FROM prospects WHERE id=1").fetchone()
        self.assertEqual(json.loads(saved[0]), body["unknowns"])
        self.assertEqual(json.loads(saved[1]), body["evidence_used"])

    def test_logs_capture_api_actions_and_redact_credentials(self):
        server.record_log("ai", "test_ai_call", details={"api_key": "do-not-store", "model": "test", "prompt": "hello"})
        status, body = self.request("POST", "/api/logs", {"action": "ui_test", "details": {"button": "test"}})
        self.assertEqual(status, 202)
        status, body = self.request("GET", "/api/logs?kind=ai")
        self.assertEqual(status, 200)
        item = next(row for row in body["items"] if row["event"] == "test_ai_call")
        self.assertEqual(item["details"]["api_key"], "[redacted]")
        self.assertEqual(item["details"]["model"], "test")
        status, body = self.request("GET", "/api/logs?kind=action")
        self.assertEqual(status, 200)
        self.assertTrue(any(row["event"] == "ui_test" for row in body["items"]))

    def test_runtime_settings_persist_and_change_live_limits(self):
        status, body = self.request("POST", "/api/settings", {
            "browser_use": {"enabled": True, "max_steps": 22, "timeout_seconds": 900, "llm_timeout_seconds": 120},
            "maps": {"request_timeout_seconds": 1800, "poll_seconds": 4, "max_depth": 3},
            "limits": {"website_scan_timeout_seconds": 15},
        })
        self.assertEqual(status, 200)
        self.assertEqual(body["browser_use"]["timeout_seconds"], 900)
        self.assertEqual(body["maps"]["request_timeout_seconds"], 1800)
        self.assertEqual(body["limits"]["website_scan_timeout_seconds"], 15)
        status, settings = self.request("GET", "/api/settings")
        self.assertEqual(status, 200)
        self.assertEqual(settings["browser_use"]["max_steps"], 22)
        self.assertEqual(settings["browser_use"]["llm_timeout_seconds"], 120)
        self.assertEqual(settings["maps"]["poll_seconds"], 4.0)
        self.assertEqual(settings["maps"]["max_depth"], 3)
        self.assertEqual(server.browser_use_limits(), (22, 900))
        self.assertEqual(server.website_scan_timeout_seconds(), 15)


if __name__ == "__main__":
    unittest.main()