MarketingTool/tests/test_app.py

import json
import os
import tempfile
import threading
import time
from pathlib import Path
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.auth_required = server.AUTH_REQUIRED
        server.AUTH_REQUIRED = False
        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):
        server.AUTH_REQUIRED = self.auth_required
        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_degoog_search_normalises_candidates_and_keeps_provenance(self):
        class Response:
            def __enter__(self): return self
            def __exit__(self, *args): return None
            def read(self, _limit=None):
                return json.dumps({"results": [{"url": "https://www.acme.co.za/", "title": "Acme Plumbing", "content": "Acme Plumbing Cape Town"}]}).encode()
        prospect = {"name": "Acme Plumbing", "address": "Cape Town", "category": "Plumbers", "phone": ""}
        with patch.dict(os.environ, {"DEGOOG_ENABLED": "true", "DEGOOG_URL": "http://degoog:4444"}, clear=False), patch("api.browser_investigator.urllib.request.urlopen", return_value=Response()):
            results = browser_investigator._degoog_search_candidates(prospect, ['"Acme Plumbing" Cape Town'], requested=3, timeout_seconds=10)
        self.assertEqual(results[0]["source"], "degoog")
        self.assertEqual(results[0]["domain"], "acme.co.za")

        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_deepseek_provider_uses_fixed_endpoint_and_default_models(self):
        status, body = self.request("GET", "/api/ai/models?provider=deepseek")
        self.assertEqual(status, 200)
        self.assertEqual(body["base_url"], "https://api.deepseek.com/v1")
        self.assertEqual([item["id"] for item in body["models"]], ["deepseek-chat", "deepseek-reasoner"])
        status, body = self.request("POST", "/api/settings", {"ai": {"AI_PROVIDER": "deepseek", "AI_API_KEY": "deepseek-test-key"}})
        self.assertEqual(status, 200)
        self.assertEqual(body["ai"]["base_url"], "https://api.deepseek.com/v1")
        self.assertEqual(body["ai"]["model"], "deepseek-chat")
        self.assertTrue(body["ai"]["configured"])
        status, body = self.request("GET", "/api/config")
        self.assertEqual(status, 200)
        self.assertEqual(body["ai_provider"], "deepseek")
        self.assertNotIn("deepseek-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://api.deepseek.com/v1/chat/completions"))
            self.assertEqual(http_json.call_args.args[2]["thinking"], {"type": "disabled"})

    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_prospect_workspace_has_pipeline_and_view_switcher(self):
        script = Path(__file__).resolve().parents[1].joinpath("web", "app.js").read_text()
        self.assertIn("prospectBoard", script)
        self.assertIn("prospectViewMode", script)
        self.assertIn("data-board-status", script)
        self.assertIn("Needs review", script)

    def test_firecrawl_search_normalises_public_results(self):
        response = {"success": True, "data": {"web": [{"title": "Cape Town Plumbing", "url": "https://plumber.example.co.za/", "description": "Local plumbing services"}]}}
        with patch("api.server.http_json", return_value=response) as request, patch("api.server._safe_url", return_value=True):
            results = server.firecrawl_search("plumbers Cape Town", 5, api_key="fc-test")
        self.assertEqual(results[0]["website"], "https://plumber.example.co.za")
        self.assertEqual(results[0]["source"], "firecrawl")
        self.assertEqual(results[0]["description"], "Local plumbing services")
        self.assertEqual(request.call_args.args[0], "POST")
        self.assertIn("/v2/search", request.call_args.args[1])
        self.assertEqual(request.call_args.kwargs["headers"]["Authorization"], "Bearer fc-test")

    def test_firecrawl_backend_requires_key_and_is_exposed_without_secret(self):
        with patch.dict(os.environ, {"DISCOVERY_BACKEND": "firecrawl", "FIRECRAWL_API_KEY": ""}, clear=False):
            with self.assertRaisesRegex(RuntimeError, "firecrawl_api_key_required"):
                server.firecrawl_search("plumbers Cape Town", 5)

    def test_firecrawl_settings_persist_without_returning_secret(self):
        status, result = self.request("POST", "/api/settings", {"discovery": {"backend": "firecrawl", "firecrawl_api_key": "fc-secret-test"}})
        self.assertEqual(status, 200)
        self.assertTrue(result["discovery"]["firecrawl_configured"])
        self.assertNotIn("fc-secret-test", json.dumps(result))
        status, settings = self.request("GET", "/api/settings")
        self.assertEqual(status, 200)
        self.assertEqual(settings["discovery"]["backend"], "firecrawl")
        self.assertNotIn("fc-secret-test", json.dumps(settings))

    def test_live_site_snapshot_captures_design_signals_for_generation(self):
        class Response:
            headers = {"Content-Type": "text/html; charset=utf-8"}
            def __enter__(self): return self
            def __exit__(self, *args): return None
            def read(self, limit): return b'''<html class="brand-site"><head><title>Acme</title><style>:root{--brand:#123456} body{font-family: Georgia}</style></head><body><nav><a href="/services">Services</a></nav><main><h1>Acme Plumbing</h1><section class="hero dark"><img src="https://example.com/hero.jpg"><h2>Fast help</h2></section></main></body></html>'''
            def geturl(self): return "https://acme.example/"
        with patch("api.server._safe_url", return_value=True), patch("api.server.urllib.request.urlopen", return_value=Response()):
            snapshot = server.fetch_company_page("https://acme.example/")
        self.assertIn("hero", snapshot["website_design_brief"])
        self.assertIn("#123456", snapshot["website_design_brief"])
        self.assertIn("Georgia", snapshot["website_design_brief"])
        self.assertIn("https://example.com/hero.jpg", snapshot["website_design_brief"])

    def test_rendered_design_variants_are_not_the_same_template(self):
        prospect = {"name": "Acme Plumbing", "category": "plumbers", "address": "Cape Town"}
        first = server.premium_render_pages(prospect, {}, mode="onepage", design={"variant": "midnight", "layout": "editorial", "radius": "sharp"})["home"]
        second = server.premium_render_pages(prospect, {}, mode="onepage", design={"variant": "warm", "layout": "bento", "radius": "soft"})["home"]
        self.assertNotEqual(first, second)
        self.assertIn("layout-editorial", first)
        self.assertIn("layout-bento", second)

    def test_mockup_fallback_is_saved_and_public_share_link_works(self):
        conn = server.db()
        conn.execute("INSERT INTO prospects(name,domain,category,address,phone,emails,website,website_status,discovered_at) VALUES(?,?,?,?,?,?,?,?,?)", ("Acme Plumbing", "maps:acme", "plumbers", "Cape Town", "+27101234567", "info@acme.co.za", "", "no_detected_website", server.now()))
        conn.commit(); conn.close()
        status, created = self.request("POST", "/api/prospects/1/mockup", {"instructions": "Keep it friendly"})
        self.assertEqual(status, 201)
        self.assertIn("/mockups/", created["share_url"])
        self.assertEqual(created["generation_source"], "fallback")
        status, listed = self.request("GET", "/api/prospects/1/mockups")
        self.assertEqual(status, 200)
        self.assertEqual(listed["items"][0]["id"], created["id"])
        preview_path = "/mockups/" + created["token"]
        self.conn.request("GET", preview_path)
        response = self.conn.getresponse()
        body = response.read().decode("utf-8")
        self.assertEqual(response.status, 200)
        self.assertIn("text/html", response.getheader("Content-Type"))
        self.assertIn("Acme Plumbing", body)
        self.assertNotIn("<script", body.lower())

    def test_mockup_sanitizer_removes_active_markup(self):
        html = server.sanitize_mockup_html('<html><body onload="bad"><script>alert(1)</script><a href="javascript:bad()">Hi</a></body></html>')
        self.assertIn("<html>", html)
        self.assertNotIn("script", html.lower())
        self.assertNotIn("onload", html.lower())
        self.assertNotIn("javascript:", html.lower())

    def test_multipage_mockup_exposes_real_page_routes(self):
        conn = server.db()
        conn.execute("INSERT INTO prospects(name,domain,category,discovered_at) VALUES(?,?,?,?)", ("Brightspark Electrical", "maps:bright", "electrician", server.now()))
        conn.commit(); conn.close()
        status, created = self.request("POST", "/api/prospects/1/mockup", {"mode": "multipage"})
        self.assertEqual(status, 201)
        self.assertEqual(created["mode"], "multipage")
        self.assertEqual({item["slug"] for item in created["pages"]}, {"home", "services", "about", "contact"})
        self.conn.request("GET", "/mockups/" + created["token"] + "/services")
        response = self.conn.getresponse()
        body = response.read().decode("utf-8")
        self.assertEqual(response.status, 200)
        self.assertIn("Services", body)
        self.assertIn("images.unsplash.com", body)
        self.assertIn("https://images.unsplash.com", response.getheader("Content-Security-Policy"))
        self.assertIn("frame-ancestors 'none'", response.getheader("Content-Security-Policy"))

    def test_mockup_editor_creates_a_new_version_and_keeps_original(self):
        conn = server.db()
        conn.execute("INSERT INTO prospects(name,domain,category,discovered_at) VALUES(?,?,?,?)", ("Brightspark Electrical", "maps:bright-editor", "electrician", server.now()))
        conn.commit(); conn.close()
        status, original = self.request("POST", "/api/prospects/1/mockup", {"mode": "multipage"})
        self.assertEqual(status, 201)
        self.assertEqual(original.get("version"), None)  # v1 rows remain backwards-compatible
        conn = server.db()
        for key, value in (("AI_PROVIDER", "nous"), ("AI_MODEL", "meituan/longcat-2.0:free"), ("AI_API_KEY", "test-key")):
            conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", (key, value, server.now()))
        conn.commit(); conn.close()
        ai_response = {"choices": [{"message": {"content": json.dumps({
            "tagline": "Powerful work. Presented with confidence.",
            "intro": "A sharper digital home for Brightspark Electrical.",
            "services": [{"name": "Electrical services", "description": "A clear way to understand the work available."}],
            "about": "A refined concept based on the public business listing.",
            "cta": "Let’s make the next step easy.",
            "hero_image": server.STOCK_IMAGE_LIBRARY["electric"][0],
            "gallery_image": server.STOCK_IMAGE_LIBRARY["electric"][1],
        })}}]}
        with patch("api.server.http_json", return_value=ai_response) as request:
            status, edited = self.request("POST", f"/api/mockups/{original['id']}/edit", {"instruction": "Make the whole site darker and put more emphasis on services.", "page": "all"})
        self.assertEqual(status, 201)
        self.assertEqual(edited["version"], 2)
        self.assertEqual(edited["parent_id"], original["id"])
        self.assertEqual(edited["mode"], "multipage")
        self.assertIn("Applied your request", edited["change_summary"])
        self.assertIn("max_tokens", request.call_args.args[2])
        self.assertEqual(request.call_args.args[2]["max_tokens"], 1800)
        self.conn.request("GET", "/mockups/" + edited["token"] + "/services?embed=1")
        response = self.conn.getresponse(); body = response.read().decode("utf-8")
        self.assertEqual(response.status, 200)
        self.assertIn("frame-ancestors 'self'", response.getheader("Content-Security-Policy"))
        self.assertIn("--paper:#091321", body)
        self.assertIn("Electrical services", body)
        self.conn.request("GET", "/mockups/" + edited["token"])
        response = self.conn.getresponse(); edited_home = response.read().decode("utf-8")
        self.assertEqual(response.status, 200)
        self.assertIn("Powerful work", edited_home)
        self.conn.request("GET", "/mockups/" + original["token"] + "/services")
        response = self.conn.getresponse(); original_body = response.read().decode("utf-8")
        self.assertEqual(response.status, 200)
        self.assertIn("Brightspark Electrical", original_body)
        status, metadata = self.request("GET", f"/api/mockups/{edited['id']}")
        self.assertEqual(status, 200)
        self.assertEqual(metadata["version"], 2)
        status, listed = self.request("GET", "/api/prospects/1/mockups")
        self.assertEqual(status, 200)
        self.assertEqual([item["version"] for item in listed["items"]], [2, 1])

    def test_multipage_ai_uses_compact_content_and_renders_all_pages(self):
        conn = server.db()
        conn.execute("INSERT INTO prospects(name,domain,category,discovered_at) VALUES(?,?,?,?)", ("Brightspark Electrical", "maps:bright-ai", "electrician", server.now()))
        for key, value in (("AI_PROVIDER", "nous"), ("AI_MODEL", "meituan/longcat-2.0:free"), ("AI_API_KEY", "test-key")):
            conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", (key, value, server.now()))
        conn.commit(); conn.close()
        response = {"choices": [{"message": {"content": json.dumps({
            "tagline": "Powerful work. Presented with clarity.",
            "intro": "A considered digital home for Brightspark Electrical.",
            "services": [{"name": "Electrical services", "description": "A clear way to understand the work available."}],
            "about": "A refined concept grounded in the public business listing.",
            "cta": "Let’s make the next step easy.",
            "hero_image": server.STOCK_IMAGE_LIBRARY["electric"][0],
            "gallery_image": server.STOCK_IMAGE_LIBRARY["electric"][1],
        })}}]}
        with patch("api.server.http_json", return_value=response) as request:
            status, created = self.request("POST", "/api/prospects/1/mockup", {"mode": "multipage"})
        self.assertEqual(status, 201)
        self.assertEqual(created["generation_source"], "ai")
        self.assertEqual({item["slug"] for item in created["pages"]}, {"home", "services", "about", "contact"})
        payload = request.call_args.args[2]
        self.assertEqual(payload["max_tokens"], 1800)
        self.assertNotIn("pages", json.dumps(payload["messages"]))
        stored = server.db().execute("SELECT pages_json FROM mockups WHERE id=?", (created["id"],)).fetchone()
        pages = json.loads(stored[0])
        self.assertIn("Powerful work", pages["home"])
        self.assertTrue(all("Brightspark Electrical" in html and "@keyframes" in html and "<footer" in html for html in pages.values()))

    def test_onepage_mockup_uses_in_page_navigation(self):
        html = server.premium_render_pages({"name": "Acme", "category": "plumber"}, mode="onepage")["home"]
        self.assertIn('href="#services"', html)
        self.assertIn('id="services"', html)
        self.assertIn('id="about"', html)
        self.assertIn('id="contact"', html)

    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_direct_search_scoring_recognises_compact_business_domain(self):
        prospect = {"name": "BRIGHTSPARK ELECTRICAL Services", "address": '{"city":"Cape Town"}', "phone": "067 732 2090"}
        confidence, evidence = browser_investigator._candidate_score(
            prospect, "https://brightsparkelec.co.za/", "Brightspark Electrical Services", "Cape Town 067 732 2090"
        )
        self.assertGreaterEqual(confidence, 80)
        self.assertTrue(any("domain" in item for item in evidence))

    def test_website_candidates_are_limited_to_coza_and_com(self):
        raw = '{"candidates":[{"url":"https://bright.example.org","confidence":99},{"url":"https://brightsparkelec.co.za","confidence":80},{"url":"https://brightspark.com","confidence":75}]}'
        candidates = browser_investigator._parse_agent_result(raw)
        self.assertEqual([item["domain"] for item in candidates], ["brightsparkelec.co.za", "brightspark.com"])

    def test_direct_search_rejects_generic_result_without_identity_signal(self):
        prospect = {"name": "BRIGHTSPARK ELECTRICAL Services", "address": '{"city":"Cape Town"}'}
        confidence, evidence = browser_investigator._candidate_score(
            prospect, "https://unrelated-electrical.co.za/", "Electrical services in Cape Town", "Directory listing"
        )
        self.assertEqual(confidence, 0)
        self.assertEqual(evidence, [])

    def test_dom_search_candidates_are_saved_before_browser_agent(self):
        conn = server.db()
        conn.execute("INSERT INTO prospects(name,domain,category,address,discovered_at) VALUES(?,?,?,?,?)", ("Brightspark Electrical Services", "maps:bright", "Electrician", "Cape Town", server.now()))
        conn.commit(); conn.close()
        found = [{"url": "https://brightsparkelec.co.za/", "domain": "brightsparkelec.co.za", "confidence": 88, "evidence": ["visible on bing search results"], "source": "browser_dom"}]
        with patch("api.server.direct_search_candidates", return_value=found) as direct, patch("api.server.investigate_websites") as agent:
            status, body = self.request("POST", "/api/prospects/1/website-candidates", {})
        self.assertEqual(status, 200)
        self.assertEqual(body["source"], "browser_dom")
        self.assertEqual(body["items"][0]["domain"], "brightsparkelec.co.za")
        self.assertEqual(body["items"][0]["source"], "browser_dom")
        direct.assert_called_once()
        agent.assert_not_called()

    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}}]}) as http_json:
            status, body = self.request("POST", "/api/prospects/1/message", {"offer": "Gold Plus hosting at R119/month with a free one-page website.", "tone": "consultative", "goal": "review"})
        self.assertEqual(status, 200)
        request_payload = http_json.call_args.args[2]
        prompt = json.loads(request_payload["messages"][1]["content"])
        self.assertIn('"tone": "consultative"', request_payload["messages"][1]["content"])
        self.assertIn("client_specific_strategy", prompt["evidence"])
        self.assertIn("private planning notes", request_payload["messages"][0]["content"])
        self.assertIn("Do not mention Google", request_payload["messages"][0]["content"])
        self.assertEqual(request_payload["temperature"], 0.65)
        self.assertIn("consultative and helpful", request_payload["messages"][0]["content"])
        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_message_style_pass_removes_stock_openers_without_rewriting_facts(self):
        text = server._clean_generated_message(
            "Good day Acme Plumbing,\n\nI hope you are well. I noticed your business listing does not show a website.\n\nKind regards",
            "Acme Plumbing",
        )
        self.assertTrue(text.startswith("Hi Acme Plumbing,"))
        self.assertNotIn("I hope you are well", text)
        self.assertNotIn("I noticed your business", text)
        self.assertIn("does not show a website", text)
        self.assertIn("Kind regards", text)

    def test_message_strategy_uses_concise_greeting_and_missing_site_shape(self):
        prospect = {
            "id": 7,
            "name": "Ash Electrical Cape Town",
            "address": "Salt River, Cape Town",
            "search_category": "Electricians",
            "website_status": "no_detected_website",
        }
        self.assertEqual(server._message_display_name(prospect), "Ash Electrical")
        strategy = server._message_strategy_context(prospect)
        self.assertEqual(strategy["business_name_for_greeting"], "Ash Electrical")
        self.assertIn("looking at local Electricians in Cape Town", strategy["preferred_message_shape"])
        self.assertIn("dedicated website", strategy["preferred_message_shape"])

    def test_email_draft_can_include_latest_mockup_link(self):
        conn = server.db()
        conn.execute("INSERT INTO prospects(name,domain,category,discovered_at) VALUES(?,?,?,?)", ("Acme", "maps:acme", "plumber", server.now()))
        conn.execute("INSERT INTO mockups(organization_id,prospect_id,token,title,html,created_at,updated_at) VALUES(?,?,?,?,?,?,?)", (1, 1, "a" * 24, "Acme · Website concept", "<html><body>Acme</body></html>", server.now(), 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 concept for Acme", "message": "Good day Acme,\n\nWe can help.", "rationale": "Uses the supplied record.", "recommended_offer": "Website", "unknowns": [], "evidence_used": ["prospect_record"], "follow_up": "Review before sending."})
        with patch("api.server.http_json", return_value={"choices": [{"message": {"content": content}}]}):
            status, body = self.request("POST", "/api/prospects/1/message", {"offer": "A modern website"})
        self.assertEqual(status, 200)
        self.assertIn("/mockups/", body["message"])
        self.assertIn("mockup_preview", body["evidence_used"])
        self.assertTrue(body["mockup_url"])

    def test_email_draft_chat_revision_uses_current_text_and_persists(self):
        conn = server.db()
        conn.execute(
            "INSERT INTO prospects(name,domain,search_category,category,address,maps_url,website_status,emails,opportunity_score,ai_subject,ai_message,ai_offer,discovered_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)",
            ("Acme Plumbing", "maps:chat", "plumbers", "Plumber", "Cape Town", "https://maps.example/acme", "no_detected_website", "acme@gmail.com", 88,
             "A practical website idea for Acme", "Good day Acme Plumbing,\n\nYour public listing does not show a dedicated website. We can help.\n\nKind regards", "Gold Plus hosting at R119/month with a free rebuild.", 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 softer website idea for Acme",
            "message": "Good day Acme Plumbing,\n\nYour public listing does not show a dedicated website. We can help with Gold Plus hosting at R119/month and a free rebuild. Would a short chat be useful?\n\nKind regards",
            "rationale": "Shortened the opening and softened the call to action as requested.",
            "recommended_offer": "Gold Plus hosting at R119/month with a free rebuild.",
            "unknowns": ["Confirm the preferred contact person"],
            "evidence_used": ["prospect_record", "public_maps_listing", "public_email_listing"],
            "follow_up": "Check the wording and offer before sending.",
        })
        with patch("api.server.http_json", return_value={"choices": [{"message": {"content": content}}]}) as http_json:
            status, body = self.request("POST", "/api/prospects/1/message/edit", {
                "instruction": "Make it warmer, shorter, and use a softer call to action.",
                "current_subject": "A practical website idea for Acme",
                "current_message": "Good day Acme Plumbing,\n\nYour public listing does not show a dedicated website. We can help.\n\nKind regards",
                "offer": "Gold Plus hosting at R119/month with a free rebuild.",
                "tone": "warm",
                "goal": "conversation",
                "include_mockup": False,
            })
        self.assertEqual(status, 200)
        self.assertEqual(body["subject"], "A softer website idea for Acme")
        self.assertIn("softened", body["rationale"])
        request_payload = http_json.call_args.args[2]
        prompt = json.loads(request_payload["messages"][1]["content"])
        self.assertEqual(prompt["current_draft"]["subject"], "A practical website idea for Acme")
        self.assertIn("Make it warmer", prompt["edit_instruction"])
        self.assertEqual(request_payload["max_tokens"], 1000)
        self.assertIn("cannot add facts", request_payload["messages"][0]["content"])
        saved = server.db().execute("SELECT ai_subject,ai_message,ai_offer FROM prospects WHERE id=1").fetchone()
        self.assertEqual(saved[0], body["subject"])
        self.assertEqual(saved[1], body["message"])
        self.assertEqual(saved[2], "Gold Plus hosting at R119/month with a free rebuild.")
        status, logs = self.request("GET", "/api/logs?kind=ai")
        self.assertEqual(status, 200)
        self.assertTrue(any(item["event"] == "message_revision_completed" for item in logs["items"]))

    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, "ai_mockup_timeout_seconds": 75},
        })
        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)
        self.assertEqual(body["limits"]["ai_mockup_timeout_seconds"], 75)
        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(settings["limits"]["ai_mockup_timeout_seconds"], 75)
        self.assertEqual(server.browser_use_limits(), (22, 900))
        self.assertEqual(server.website_scan_timeout_seconds(), 15)

    def test_company_profile_and_promotions_feed_email_draft(self):
        status, profile = self.request("POST", "/api/settings", {"company_profile": {
            "name": "Marco Web Studio", "description": "A South African hosting studio.",
            "services": "Domains, hosting and website rebuilds.", "packages": "Gold Plus hosting at R119/month.",
        }})
        self.assertEqual(status, 200)
        self.assertTrue(profile["company_profile"]["configured"])
        self.request("POST", "/api/settings", {"ai": {"AI_PROVIDER": "custom", "AI_BASE_URL": "https://ai.example/v1", "AI_MODEL": "test-model", "AI_API_KEY": "test-key"}})
        status, promotion = self.request("POST", "/api/promotions", {"name": "Free rebuild", "description": "Free website rebuild when hosting moves to Gold Plus.", "terms": "New hosting customers only."})
        self.assertEqual(status, 201)
        status, listed = self.request("GET", "/api/promotions")
        self.assertEqual(status, 200)
        self.assertEqual(listed["items"][0]["name"], "Free rebuild")
        conn = server.db()
        conn.execute("INSERT INTO prospects(name,domain,category,discovered_at) VALUES(?,?,?,?)", ("Acme", "maps:acme", "plumber", server.now()))
        conn.commit(); conn.close()
        content = json.dumps({"subject": "A free website rebuild", "message": "Good day Acme,\n\nWe can help.", "rationale": "Uses the supplied promotion.", "recommended_offer": "Free rebuild", "unknowns": [], "evidence_used": ["prospect_record"], "follow_up": "Review before sending."})
        with patch("api.server.http_json", return_value={"choices": [{"message": {"content": content}}]}) as http_json:
            status, body = self.request("POST", "/api/prospects/1/message", {"offer": "Please mention our offer.", "promotion_id": promotion["id"]})
        self.assertEqual(status, 200)
        prompt = http_json.call_args.args[2]["messages"][1]["content"]
        self.assertIn("Free website rebuild", prompt)
        self.assertIn("Gold Plus hosting at R119/month", prompt)
        saved = server.db().execute("SELECT ai_promotion_id FROM prospects WHERE id=1").fetchone()
        self.assertEqual(saved[0], promotion["id"])

    @patch("api.server.ai_assess_prospect", return_value={"ai_probability": 88, "ai_recommendation": "contact", "ai_reasoning": "Active business with a broken site."})
    def test_workspace_workflow_and_assessment_history(self, assess):
        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": "Cape Town"})
            self.wait_for_run(run["id"])
        conn = server.db(); conn.execute("INSERT INTO users(organization_id,email,name,password_hash,role,active,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?)", (1, "reviewer@example.com", "Reviewer", "", "reviewer", 1, server.now(), server.now())); conn.commit(); user_id = conn.execute("SELECT id FROM users WHERE email='reviewer@example.com'").fetchone()[0]; conn.close()
        status, note = self.request("POST", "/api/prospects/1/notes", {"body": "Call after reviewing the website."}); self.assertEqual(status, 201)
        status, task = self.request("POST", "/api/prospects/1/tasks", {"title": "Send tailored draft", "due_at": "2030-01-01"}); self.assertEqual(status, 201)
        status, assigned = self.request("POST", "/api/prospects/1/assignment", {"assigned_user_id": user_id}); self.assertEqual(status, 200); self.assertEqual(assigned["assigned_user_id"], user_id)
        status, body = self.request("POST", "/api/prospects/1/assess", {}); self.assertEqual(status, 200); self.assertEqual(body["ai_probability"], 88)
        status, history = self.request("GET", "/api/prospects/1/history"); self.assertEqual(status, 200); self.assertTrue(history["score_history"]); self.assertTrue(history["ai_history"])
        status, tasks = self.request("GET", "/api/prospects/1/tasks"); self.assertEqual(status, 200); self.assertEqual(tasks["items"][0]["title"], "Send tailored draft")


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