+
+
+
+ """
+ out = analyze_html(
+ html=html,
+ page_url="https://site.com/page",
+ base_url="https://site.com/page",
+ canonical_url="",
+ )
+ assert out["internal_link_count"] >= 1
+ assert out["external_link_count"] >= 1
+ assert out["html_lang"] == "en"
+ assert isinstance(out["warnings"], list)
+ json.dumps(out)
+
+
+def test_json_ld_missing_type_detection() -> None:
+ from website_profiling.analysis.page import _json_ld_missing_type
+
+ assert _json_ld_missing_type({"@context": "x"}) is False
+ assert _json_ld_missing_type({"name": "Acme"}) is True
+ assert _json_ld_missing_type({"@graph": [{"@type": "Thing"}, {"name": "X"}]}) is True
+
diff --git a/tests/test_cli_dispatch.py b/tests/test_cli_dispatch.py
new file mode 100644
index 00000000..646b1c14
--- /dev/null
+++ b/tests/test_cli_dispatch.py
@@ -0,0 +1,21 @@
+"""Smoke tests for CLI argument parsing."""
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_cli_help_lists_commands():
+ proc = subprocess.run(
+ [sys.executable, "-m", "src", "--help"],
+ cwd=REPO_ROOT,
+ capture_output=True,
+ text=True,
+ timeout=15,
+ )
+ assert proc.returncode == 0
+ for cmd in ("crawl", "report", "plot", "lighthouse", "keywords", "warnings", "enrich", "google"):
+ assert cmd in proc.stdout
diff --git a/tests/test_commands_wrappers_unit.py b/tests/test_commands_wrappers_unit.py
new file mode 100644
index 00000000..1211d793
--- /dev/null
+++ b/tests/test_commands_wrappers_unit.py
@@ -0,0 +1,83 @@
+import argparse
+import types
+
+import pytest
+
+
+def test_google_cmd_list_properties(monkeypatch, capsys) -> None:
+ from website_profiling.commands import google_cmd
+
+ # Fake integrations.google.fetch.list_properties
+ fake_fetch = types.SimpleNamespace(list_properties=lambda _p=None: {"ok": True}, fetch_google_data=None)
+ fake_auth = types.SimpleNamespace(build_credentials=None, read_secrets=None)
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.fetch",
+ fake_fetch,
+ )
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.auth",
+ fake_auth,
+ )
+
+ args = argparse.Namespace(list_properties=True, test=False)
+ with pytest.raises(SystemExit) as e:
+ google_cmd.run({"google_credentials_path": ""}, cwd="/tmp", path=lambda k, d: d, args=args)
+ assert e.value.code == 0
+ out = capsys.readouterr().out.strip()
+ assert out == '{"ok": true}' or out == '{"ok": True}'
+
+
+def test_keywords_cmd_expand_only(monkeypatch, capsys) -> None:
+ from website_profiling.commands import keywords_cmd
+
+ def fake_expand(seeds, sources):
+ return {"seeds": seeds, "sources": list(sources)}
+
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.suggest",
+ types.SimpleNamespace(batch_expand=fake_expand),
+ )
+ args = argparse.Namespace(expand_only=True, enrich_google=False)
+ with pytest.raises(SystemExit) as e:
+ keywords_cmd.run({"keyword_seeds": "a,b"}, args)
+ assert e.value.code == 0
+ assert '"seeds": ["a", "b"]' in capsys.readouterr().out
+
+
+def test_warnings_cmd_run(monkeypatch) -> None:
+ from website_profiling.commands import warnings_cmd
+
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.tools.warnings",
+ types.SimpleNamespace(main=lambda **_kw: 0),
+ )
+ args = argparse.Namespace()
+ with pytest.raises(SystemExit) as e:
+ warnings_cmd.run({"warning_mapper_input": "", "warning_mapper_input_type": "lighthouse"}, cwd="/tmp", path=lambda k, d: d, args=args)
+ assert e.value.code == 0
+
+
+def test_lighthouse_cmd_run_calls_runner(monkeypatch) -> None:
+ from website_profiling.commands import lighthouse_cmd
+
+ called = {"n": 0}
+
+ def fake_main(**_kwargs):
+ called["n"] += 1
+ return 0
+
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.lighthouse.runner",
+ types.SimpleNamespace(main=fake_main),
+ )
+ args = argparse.Namespace()
+ with pytest.raises(SystemExit) as e:
+ lighthouse_cmd.run({"lighthouse_url": "https://x", "start_url": "https://x"}, args)
+ assert e.value.code == 0
+ assert called["n"] == 1
+
diff --git a/tests/test_common_io.py b/tests/test_common_io.py
new file mode 100644
index 00000000..ec266b46
--- /dev/null
+++ b/tests/test_common_io.py
@@ -0,0 +1,53 @@
+import pandas as pd
+
+
+def test_load_save_dataframe_and_edges(tmp_path) -> None:
+ from website_profiling.common import load_dataframe, load_edges, save_dataframe, save_edges
+
+ df = pd.DataFrame([{"a": 1}, {"a": 2}])
+ p = tmp_path / "x.csv"
+ save_dataframe(df, str(p))
+ df2 = load_dataframe(str(p))
+ assert df2.shape[0] == 2
+
+ edges = [("a", "b"), ("b", "c")]
+ ep = tmp_path / "edges.csv"
+ save_edges(edges, str(ep))
+ edges2 = load_edges(str(ep))
+ assert edges2 == edges
+
+
+def test_parse_resources_and_social_meta() -> None:
+ from bs4 import BeautifulSoup
+
+ from website_profiling.common import parse_resources, parse_social_meta
+
+ html = """
+
+
+
+
+
+
+
+ """
+ soup = BeautifulSoup(html, "lxml")
+ social = parse_social_meta(soup)
+ assert social["og_title"] == "T"
+ assert social["twitter_card"] == "summary"
+
+ res = parse_resources(html, "https://site.com")
+ assert res["script_count"] == 1
+ assert res["link_stylesheet_count"] == 1
+
+
+def test_parse_content_text_extracts_keywords_and_excerpt() -> None:
+ from bs4 import BeautifulSoup
+
+ from website_profiling.common import parse_content_text
+
+ html = "Hello again. This is a test sentence.
" + soup = BeautifulSoup(html, "lxml") + out = parse_content_text(soup, raw_html=html, excerpt_max_chars=30) + assert out["word_count"] > 0 + assert out["content_excerpt"] diff --git a/tests/test_common_parsing.py b/tests/test_common_parsing.py new file mode 100644 index 00000000..9470e287 --- /dev/null +++ b/tests/test_common_parsing.py @@ -0,0 +1,79 @@ +import json + + +def test_normalize_link_filters_schemes_and_strips_fragment_and_slash() -> None: + from website_profiling.common import normalize_link + + assert normalize_link("https://x.com", "mailto:test@example.com") is None + assert normalize_link("https://x.com", "javascript:alert(1)") is None + assert normalize_link("https://x.com", "ftp://x.com/a") is None + + assert normalize_link("https://x.com/base/", "/a#frag") == "https://x.com/a" + assert normalize_link("https://x.com/base/", "https://x.com/a/") == "https://x.com/a" + + +def test_parse_links_and_title() -> None: + from website_profiling.common import parse_links + + html = """ +
+ 
+
+
+ """
+ meta_desc, meta_len, h1_text, h1_count, canon = parse_seo("https://s.com", html)
+ assert meta_desc == "Desc"
+ assert meta_len == 4
+ assert h1_text == "H1"
+ assert h1_count == 1
+ assert canon == "https://s.com/canon"
+
+ ext = parse_seo_extended(html, "https://s.com")
+ assert ext["viewport_present"] is True
+ assert ext["noindex"] is True
+ assert ext["has_schema"] is True
+ assert ext["images_total"] == 2
+ assert ext["images_without_alt"] == 1
+ assert ext["mixed_content_count"] > 0
+
+ json.dumps(ext)
+
+
+def test_parse_links_serialized_and_empty_detection() -> None:
+ from website_profiling.common import _is_empty, parse_links_serialized
+
+ assert _is_empty(None) is True
+ assert _is_empty("") is True
+ assert _is_empty([]) is False
+ assert _is_empty(["x"]) is False
+
+ assert parse_links_serialized('["a","b"]') == ["a", "b"]
+ assert parse_links_serialized("") == []
+ assert parse_links_serialized(None) == []
+
diff --git a/tests/test_config_parsing_unit.py b/tests/test_config_parsing_unit.py
new file mode 100644
index 00000000..22170125
--- /dev/null
+++ b/tests/test_config_parsing_unit.py
@@ -0,0 +1,38 @@
+import os
+
+
+def test_load_config_parses_equals_and_colon_and_ignores_comments(tmp_path) -> None:
+ from website_profiling.config import load_config
+
+ p = tmp_path / "cfg.txt"
+ p.write_text(
+ "\n".join(
+ [
+ "# comment",
+ "start_url = https://example.com",
+ "site_name: Example",
+ "badline",
+ "",
+ ]
+ ),
+ encoding="utf-8",
+ )
+ cfg = load_config(str(p))
+ assert cfg["start_url"] == "https://example.com"
+ assert cfg["site_name"] == "Example"
+ assert "badline" not in cfg
+
+
+def test_getters_bool_int_float_list() -> None:
+ from website_profiling.config import get_bool, get_float, get_int, get_list
+
+ cfg = {"b1": "true", "b2": "0", "i": "10", "f": "1.25", "l": " a, b , ,c "}
+ assert get_bool(cfg, "b1", False) is True
+ assert get_bool(cfg, "b2", True) is False
+ assert get_int(cfg, "i") == 10
+ assert get_int(cfg, "missing", 7) == 7
+ assert get_int(cfg, "bad", 3) == 3
+ assert get_float(cfg, "f") == 1.25
+ assert get_float(cfg, "badf", 2.5) == 2.5
+ assert get_list(cfg, "l") == ["a", "b", "c"]
+
diff --git a/tests/test_config_resolution.py b/tests/test_config_resolution.py
new file mode 100644
index 00000000..3370543e
--- /dev/null
+++ b/tests/test_config_resolution.py
@@ -0,0 +1,43 @@
+"""CLI config resolution order."""
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_cli_exits_when_no_config_and_empty_db(tmp_path, monkeypatch):
+ monkeypatch.setenv("DATABASE_URL", "postgres://u:p@127.0.0.1:59999/nodb")
+ monkeypatch.setenv("DATA_DIR", str(tmp_path))
+ proc = subprocess.run(
+ [sys.executable, "-m", "src", "crawl"],
+ cwd=REPO_ROOT,
+ capture_output=True,
+ text=True,
+ timeout=30,
+ )
+ assert proc.returncode != 0
+ combined = proc.stderr + proc.stdout
+ assert "No pipeline config found" in combined or "Could not load pipeline_config" in combined
+
+
+def test_cli_loads_config_file_override(tmp_path):
+ cfg = tmp_path / "custom.txt"
+ cfg.write_text("start_url = https://example.com\nsite_name = Example\n", encoding="utf-8")
+ env = os.environ.copy()
+ env["DATABASE_URL"] = "postgres://u:p@127.0.0.1:59999/nodb"
+ proc = subprocess.run(
+ [sys.executable, "-m", "src", "--config", str(cfg), "crawl"],
+ cwd=REPO_ROOT,
+ capture_output=True,
+ text=True,
+ timeout=15,
+ env=env,
+ )
+ # Should get past config load; may fail on crawl/network but not on missing config
+ combined = proc.stderr + proc.stdout
+ assert "Config file not found" not in combined
+ assert "No pipeline config found" not in combined
diff --git a/tests/test_config_resolve_more.py b/tests/test_config_resolve_more.py
new file mode 100644
index 00000000..9f649aee
--- /dev/null
+++ b/tests/test_config_resolve_more.py
@@ -0,0 +1,37 @@
+import argparse
+
+import pytest
+
+
+def test_resolve_config_uses_explicit_file(tmp_path) -> None:
+ from website_profiling.commands.config_resolve import resolve_config
+
+ p = tmp_path / "c.txt"
+ p.write_text("start_url = https://x.com\n", encoding="utf-8")
+ args = argparse.Namespace(config=str(p))
+ cfg, cwd = resolve_config(args)
+ assert cfg["start_url"] == "https://x.com"
+ assert cwd == str(tmp_path)
+
+
+def test_make_path_fn_makes_absolute(tmp_path) -> None:
+ from website_profiling.commands.config_resolve import make_path_fn
+
+ cfg = {"p": "rel.txt"}
+ fn = make_path_fn(cfg, str(tmp_path))
+ assert fn("p", "x").startswith(str(tmp_path))
+
+
+def test_should_enrich_keywords_after_report_prefers_explicit_flag() -> None:
+ from website_profiling.commands.config_resolve import should_enrich_keywords_after_report
+
+ assert should_enrich_keywords_after_report({"enrich_keywords_after_report": "true"}) is True
+ assert should_enrich_keywords_after_report({"enrich_keywords_after_report": "false", "enable_google_search_console": "true"}) is False
+
+
+def test_resolved_lighthouse_url_falls_back_to_start_url() -> None:
+ from website_profiling.commands.config_resolve import resolved_lighthouse_url
+
+ assert resolved_lighthouse_url({"start_url": "https://x.com"}) == "https://x.com"
+ assert resolved_lighthouse_url({"lighthouse_url": "https://lh.com", "start_url": "https://x.com"}) == "https://lh.com"
+
diff --git a/tests/test_config_resolve_unit.py b/tests/test_config_resolve_unit.py
new file mode 100644
index 00000000..4fbf6688
--- /dev/null
+++ b/tests/test_config_resolve_unit.py
@@ -0,0 +1,36 @@
+import argparse
+import os
+
+import pytest
+from pathlib import Path
+
+
+def test_cleanup_lighthouse_work_dir_only_deletes_under_tmp(tmp_path, monkeypatch) -> None:
+ from website_profiling.commands.config_resolve import cleanup_lighthouse_work_dir
+
+ # Under tmp -> removed
+ d = tmp_path / "wp-lighthouse-x"
+ d.mkdir()
+ cleanup_lighthouse_work_dir(str(d))
+ assert not d.exists()
+
+ # Outside tmp -> should not remove
+ outside = Path.cwd() / "outside-wp-test"
+ outside.mkdir(exist_ok=True)
+ try:
+ cleanup_lighthouse_work_dir(str(outside))
+ assert outside.exists()
+ finally:
+ # cleanup in case it wasn't removed
+ if outside.exists():
+ outside.rmdir()
+
+
+def test_require_start_url_exits_when_missing(capsys) -> None:
+ from website_profiling.commands.config_resolve import require_start_url
+
+ with pytest.raises(SystemExit) as e:
+ require_start_url({}, for_step="crawl")
+ assert e.value.code == 1
+ assert "start_url is required" in capsys.readouterr().err
+
diff --git a/tests/test_config_schema_keys.py b/tests/test_config_schema_keys.py
new file mode 100644
index 00000000..c021739f
--- /dev/null
+++ b/tests/test_config_schema_keys.py
@@ -0,0 +1,88 @@
+"""Example config keys must match the documented schema set."""
+from __future__ import annotations
+
+from pathlib import Path
+
+from tests.config_test_utils import REPO_ROOT, parse_config_keys
+
+# Mirrors web/src/lib/pipelineConfigSchema.ts ALL_SCHEMA_KEYS (update when schema changes).
+SCHEMA_KEYS = {
+ "start_url",
+ "max_pages",
+ "concurrency",
+ "timeout",
+ "max_depth",
+ "polite_delay",
+ "ignore_robots",
+ "allow_external",
+ "store_outlinks",
+ "store_content_excerpt",
+ "content_excerpt_max_chars",
+ "preserve_crawl_history",
+ "crawl_stream_to_db",
+ "crawl_exclude_urls",
+ "outbound_domain_max_rows",
+ "include_keyword_opportunities",
+ "site_name",
+ "report_title",
+ "max_fetch_for_edges",
+ "same_domain_only",
+ "max_nodes_plot",
+ "run_security_scan",
+ "security_scan_active",
+ "security_max_urls_probe",
+ "lighthouse_url",
+ "lighthouse_mode",
+ "lighthouse_strategy",
+ "lighthouse_categories",
+ "lighthouse_iterations",
+ "run_lighthouse",
+ "run_lighthouse_on_pages",
+ "lighthouse_max_pages",
+ "lighthouse_concurrency",
+ "enable_duplicate_detection",
+ "enable_language_detection",
+ "analysis_fuzzy_threshold",
+ "analysis_simhash_hamming",
+ "analysis_dup_max_pages",
+ "run_crawl",
+ "run_report",
+ "run_plot",
+ "enable_google_search_console",
+ "enable_google_analytics",
+ "google_date_range_days",
+ "google_url_gap_list_limit",
+ "enrich_keywords_after_report",
+ "keyword_max_pages",
+ "keyword_gsc_max_rows",
+ "brand_name",
+ "keyword_seeds",
+ "enable_google_suggest",
+ "enable_google_trends",
+ "enable_wikipedia_topic",
+ "enable_datamuse",
+ "keyword_suggest_top_n",
+ "keyword_max_suggest_results",
+ "warning_mapper_input",
+ "warning_mapper_input_type",
+}
+
+
+def test_input_example_keys_are_subset_of_schema():
+ keys = parse_config_keys(REPO_ROOT / "input.txt.example")
+ extra = keys - SCHEMA_KEYS
+ assert not extra, f"Keys in input.txt.example not in schema: {sorted(extra)}"
+
+
+def test_pipeline_example_keys_are_subset_of_schema():
+ keys = parse_config_keys(REPO_ROOT / "pipeline-config.example.txt")
+ extra = keys - SCHEMA_KEYS
+ assert not extra, f"pipeline-config.example.txt has unknown keys: {sorted(extra)}"
+
+
+def test_schema_keys_documented_or_optional_in_example():
+ keys = parse_config_keys(REPO_ROOT / "pipeline-config.example.txt")
+ # Tristate 'auto' keys may be omitted from static examples (UI omits them when auto).
+ optional_omitted = {"enrich_keywords_after_report"}
+ missing = SCHEMA_KEYS - keys - optional_omitted
+ assert not missing, f"pipeline-config.example.txt missing keys: {sorted(missing)}"
diff --git a/tests/test_config_schema_sync.py b/tests/test_config_schema_sync.py
new file mode 100644
index 00000000..632f1c84
--- /dev/null
+++ b/tests/test_config_schema_sync.py
@@ -0,0 +1,21 @@
+"""Example config files should stay aligned with each other."""
+from __future__ import annotations
+
+from pathlib import Path
+
+from tests.config_test_utils import REPO_ROOT, parse_config_keys
+
+
+def test_input_and_pipeline_example_have_same_keys():
+ input_keys = parse_config_keys(REPO_ROOT / "input.txt.example")
+ pipeline_keys = parse_config_keys(REPO_ROOT / "pipeline-config.example.txt")
+ assert input_keys == pipeline_keys, (
+ f"input.txt.example and pipeline-config.example.txt differ: "
+ f"only in input={sorted(input_keys - pipeline_keys)} "
+ f"only in pipeline={sorted(pipeline_keys - input_keys)}"
+ )
+
+
+def test_example_configs_have_no_legacy_google_credentials_path():
+ keys = parse_config_keys(REPO_ROOT / "input.txt.example")
+ assert "google_credentials_path" not in keys
diff --git a/tests/test_coverage_edge_win.py b/tests/test_coverage_edge_win.py
new file mode 100644
index 00000000..e9035652
--- /dev/null
+++ b/tests/test_coverage_edge_win.py
@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import pytest
+
+
+def test_pool_get_database_url_raises_when_missing(monkeypatch):
+ from website_profiling.db.pool import get_database_url
+
+ monkeypatch.delenv("DATABASE_URL", raising=False)
+ with pytest.raises(RuntimeError):
+ get_database_url()
+
+
+def test_report_store_fallback_links_and_exception_path(monkeypatch):
+ from website_profiling.db import report_store
+
+ monkeypatch.setattr(report_store, "get_crawl_run_info", lambda _c, _rid: None)
+ domain = report_store._canonical_domain_from_report(object(), {"links": [{"url": "https://fallback.com/p"}]}) # type: ignore[arg-type]
+ assert domain == "fallback.com"
+
+ class BoomConn:
+ def execute(self, *_a, **_k):
+ raise RuntimeError("x")
+
+ assert report_store.read_report_payload(BoomConn()) is None # type: ignore[arg-type]
+
diff --git a/tests/test_crawl_db_writer_imports.py b/tests/test_crawl_db_writer_imports.py
new file mode 100644
index 00000000..ebf8014f
--- /dev/null
+++ b/tests/test_crawl_db_writer_imports.py
@@ -0,0 +1,17 @@
+import pytest
+
+
+def test_crawl_db_writer_run_does_not_import_error() -> None:
+ """
+ Regression: during the db/ split, _CrawlDbWriter.run() imported helpers from
+ website_profiling.db.storage, which no longer exported them. That only fails
+ at runtime (thread start), so we exercise .run() directly with an empty queue.
+ """
+ from website_profiling.crawl.crawler import _CrawlDbWriter
+
+ writer = _CrawlDbWriter(crawl_run_id=1, batch_size=50)
+ writer.finish() # enqueue sentinel so run() exits without touching DB
+
+ # If imports are wrong, this raises ImportError immediately.
+ writer.run()
+
diff --git a/tests/test_crawl_store_more.py b/tests/test_crawl_store_more.py
new file mode 100644
index 00000000..2f2f9956
--- /dev/null
+++ b/tests/test_crawl_store_more.py
@@ -0,0 +1,21 @@
+import pandas as pd
+
+from tests.db_test_fakes import FakeConn, FakeCursor
+
+
+def test_write_crawl_empty_clears_table_when_no_run_id() -> None:
+ from website_profiling.db.crawl_store import write_crawl
+
+ conn = FakeConn()
+ write_crawl(conn, pd.DataFrame(), crawl_run_id=None) # type: ignore[arg-type]
+ assert conn.commits == 1
+
+
+def test_read_crawl_returns_empty_df_when_no_rows() -> None:
+ from website_profiling.db.crawl_store import read_crawl
+
+ conn = FakeConn()
+ conn.set_next_cursor(FakeCursor(fetchall_value=[]))
+ df = read_crawl(conn, run_id=1) # type: ignore[arg-type]
+ assert df.empty
+
diff --git a/tests/test_crawler_deep.py b/tests/test_crawler_deep.py
new file mode 100644
index 00000000..d345f795
--- /dev/null
+++ b/tests/test_crawler_deep.py
@@ -0,0 +1,148 @@
+from __future__ import annotations
+
+import json
+import types
+
+import pandas as pd
+
+
+def test_worker_success_path_populates_many_fields(monkeypatch):
+ import website_profiling.crawl.crawler as mod
+
+ c = mod.Crawler(
+ start_url="https://site.com",
+ ignore_robots=True,
+ allow_external=False,
+ use_wappalyzer=False,
+ store_outlinks=True,
+ max_depth=2,
+ )
+ c.fetch = lambda _url: ( # type: ignore[method-assign]
+ 200,
+ "text/html",
+ "ok",
+ 12,
+ 100,
+ "https://site.com/page",
+ {"X-Robots-Tag": ""},
+ 0,
+ )
+ monkeypatch.setattr(mod, "parse_links", lambda _u, _t: ("T", {"https://site.com/a", "https://ext.com/x"}))
+ monkeypatch.setattr(mod, "parse_seo", lambda *_a, **_k: ("desc", 4, "h1", 1, "https://site.com/canon"))
+ monkeypatch.setattr(
+ mod,
+ "parse_seo_extended",
+ lambda *_a, **_k: {
+ "viewport_present": True,
+ "viewport_content": "w",
+ "noindex": False,
+ "has_schema": True,
+ "heading_sequence": ["h1"],
+ "images_without_alt": 0,
+ "images_total": 1,
+ "img_without_lazy": 0,
+ "img_without_dimensions": 0,
+ "aria_count": 0,
+ "mixed_content_count": 0,
+ },
+ )
+ monkeypatch.setattr(mod, "parse_resources", lambda *_a, **_k: {"script_count": 1, "link_stylesheet_count": 1})
+ monkeypatch.setattr(
+ mod,
+ "parse_content_text",
+ lambda *_a, **_k: {
+ "word_count": 10,
+ "reading_level": 4.2,
+ "content_html_ratio": 30.0,
+ "top_keywords": "[]",
+ "content_excerpt": "abc",
+ },
+ )
+ monkeypatch.setattr(
+ mod,
+ "parse_social_meta",
+ lambda *_a, **_k: {
+ "og_title": "og",
+ "og_description": "",
+ "og_image": "",
+ "og_type": "",
+ "twitter_card": "",
+ "twitter_title": "",
+ "twitter_image": "",
+ },
+ )
+ monkeypatch.setattr(mod, "parse_tech_stack", lambda *_a, **_k: "[]")
+ monkeypatch.setattr(mod, "analyze_html", lambda *_a, **_k: {"ok": True})
+
+ out = c.worker("https://site.com")
+ assert out["status"] == 200
+ assert out["title"] == "T"
+ assert out["meta_description"] == "desc"
+ assert out["word_count"] == 10
+ assert "outlink_targets" in out
+ assert "https://site.com/a" in json.loads(out["outlink_targets"])
+
+
+def test_crawl_runs_and_handles_done_futures(monkeypatch):
+ import website_profiling.crawl.crawler as mod
+
+ c = mod.Crawler(start_url="https://site.com", ignore_robots=True, use_wappalyzer=False, concurrency=1, max_pages=1)
+ monkeypatch.setattr(
+ c,
+ "worker",
+ lambda url: {"url": url, "status": 200, "content_type": "text/html", "title": "ok", "outlinks": 0},
+ )
+ df = c.crawl(show_progress=False)
+ assert not df.empty
+ assert "crawl_time_s" in df.columns
+
+
+def test_run_crawler_writes_csv(monkeypatch, tmp_path):
+ import website_profiling.crawl.crawler as mod
+
+ class FakeCrawler:
+ def __init__(self, **_kwargs):
+ pass
+
+ def crawl(self, **_kwargs):
+ return pd.DataFrame([{"url": "https://a.com", "status": 200}])
+
+ monkeypatch.setattr(mod, "Crawler", FakeCrawler)
+ out_file = tmp_path / "out.csv"
+ df = mod.run_crawler("https://a.com", output_db=False, output_csv=str(out_file), show_progress=False)
+ assert not df.empty
+ assert out_file.exists()
+
+
+def test_run_crawler_streaming_db_path(monkeypatch):
+ import website_profiling.crawl.crawler as mod
+
+ class FakeCrawler:
+ def __init__(self, **_kwargs):
+ pass
+
+ def crawl(self, **_kwargs):
+ return pd.DataFrame([{"url": "https://a.com", "status": 200}])
+
+ class _Ctx:
+ def __enter__(self):
+ return object()
+
+ def __exit__(self, _t, _v, _tb):
+ return False
+
+ monkeypatch.setattr(mod, "Crawler", FakeCrawler)
+ fake_db = types.SimpleNamespace(
+ backup_db_if_exists=lambda: None,
+ create_crawl_run=lambda _c, _u: 10,
+ db_session=lambda: _Ctx(),
+ read_historical_data=lambda: {},
+ restore_historical_data=lambda *_a, **_k: None,
+ )
+ fake_storage = types.SimpleNamespace(ensure_crawl_tables_cleared=lambda *_a, **_k: None)
+ monkeypatch.setitem(__import__("sys").modules, "website_profiling.db", fake_db)
+ monkeypatch.setitem(__import__("sys").modules, "website_profiling.db.storage", fake_storage)
+
+ df = mod.run_crawler("https://a.com", output_db=True, crawl_stream_to_db=True, show_progress=False)
+ assert not df.empty
+
diff --git a/tests/test_crawler_unit.py b/tests/test_crawler_unit.py
new file mode 100644
index 00000000..2b350338
--- /dev/null
+++ b/tests/test_crawler_unit.py
@@ -0,0 +1,59 @@
+def test_url_matches_exclude_prefix_and_exact() -> None:
+ from website_profiling.crawl.crawler import _url_matches_exclude
+
+ assert _url_matches_exclude("https://a.com/x", []) is False
+ assert _url_matches_exclude("https://a.com/x/", ["https://a.com/x"]) is True
+ assert _url_matches_exclude("https://a.com/x/y", ["https://a.com/x/"]) is True
+ assert _url_matches_exclude("https://a.com/other", ["https://a.com/x"]) is False
+
+
+def test_crawler_init_respects_exclude_and_same_domain() -> None:
+ from website_profiling.crawl.crawler import Crawler
+
+ c = Crawler(
+ start_url="https://site.com",
+ ignore_robots=True,
+ use_wappalyzer=False,
+ exclude_urls=["https://site.com"],
+ )
+ assert c.queue.qsize() == 0
+
+ c2 = Crawler(
+ start_url="https://site.com",
+ ignore_robots=True,
+ use_wappalyzer=False,
+ exclude_urls=["https://site.com/skip"],
+ )
+ assert c2.queue.qsize() == 1
+ assert c2.same_domain("https://site.com/a") is True
+ assert c2.same_domain("https://other.com/a") is False
+
+
+def test_worker_blocked_by_robots_returns_stub_fields() -> None:
+ from website_profiling.crawl.crawler import Crawler
+
+ c = Crawler(
+ start_url="https://site.com",
+ ignore_robots=True,
+ use_wappalyzer=False,
+ store_outlinks=True,
+ )
+ # Force block
+ c.allowed_by_robots = lambda _url: False # type: ignore[method-assign]
+ out = c.worker("https://site.com/a")
+ assert out["status"] == "blocked_by_robots"
+ assert "outlink_targets" in out
+
+
+def test_worker_fetch_error_returns_error_status() -> None:
+ from website_profiling.crawl.crawler import Crawler
+
+ c = Crawler(
+ start_url="https://site.com",
+ ignore_robots=True,
+ use_wappalyzer=False,
+ )
+ c.fetch = lambda _url: (None, None, None, None, None, None, {}, 0) # type: ignore[method-assign]
+ out = c.worker("https://site.com/a")
+ assert out["status"] == "error"
+
diff --git a/tests/test_db_common.py b/tests/test_db_common.py
new file mode 100644
index 00000000..66e6b437
--- /dev/null
+++ b/tests/test_db_common.py
@@ -0,0 +1,31 @@
+import json
+
+import pytest
+
+
+def test_parse_json_field_handles_dict_list_str_and_bad_json() -> None:
+ from website_profiling.db._common import _parse_json_field
+
+ assert _parse_json_field(None) is None
+ assert _parse_json_field({"a": 1}) == {"a": 1}
+ assert _parse_json_field([1, 2]) == [1, 2]
+ assert _parse_json_field('{"x": 2}') == {"x": 2}
+ assert _parse_json_field("not-json") == "not-json"
+
+
+def test_sanitize_for_json_converts_nan_inf_and_numpy_like_item() -> None:
+ from website_profiling.db._common import _sanitize_for_json
+
+ class N:
+ def __init__(self, v):
+ self._v = v
+
+ def item(self):
+ return self._v
+
+ out = _sanitize_for_json({"a": float("nan"), "b": float("inf"), "c": N(5)})
+ assert out == {"a": None, "b": None, "c": 5}
+
+ # must be JSON-serializable
+ json.dumps(out)
+
diff --git a/tests/test_db_stores_unit.py b/tests/test_db_stores_unit.py
new file mode 100644
index 00000000..34762e41
--- /dev/null
+++ b/tests/test_db_stores_unit.py
@@ -0,0 +1,128 @@
+import pandas as pd
+
+from tests.db_test_fakes import FakeConn, FakeCursor
+
+
+def test_read_pipeline_config_splits_known_and_unknown() -> None:
+ from website_profiling.db.config_store import read_pipeline_config
+
+ conn = FakeConn()
+ conn.set_next_cursor(
+ FakeCursor(
+ fetchall_value=[
+ {"key": "start_url", "value": "https://x", "is_unknown": False},
+ {"key": "weird_key", "value": "123", "is_unknown": True},
+ ]
+ )
+ )
+ known, unknown = read_pipeline_config(conn) # type: ignore[arg-type]
+ assert known["start_url"] == "https://x"
+ assert unknown == [{"key": "weird_key", "value": "123"}]
+
+
+def test_write_pipeline_config_writes_known_and_unknown() -> None:
+ from website_profiling.db.config_store import write_pipeline_config
+
+ conn = FakeConn()
+ write_pipeline_config( # type: ignore[arg-type]
+ conn,
+ entries={"a": "1", "b": "2"},
+ unknown_keys=[{"key": "x", "value": "y"}],
+ )
+ sqls = [s for (s, _p) in conn.executed]
+ assert any("DELETE FROM pipeline_config" in s for s in sqls)
+ assert sum("INSERT INTO pipeline_config" in s for s in sqls) >= 3
+
+
+def test_crawl_rows_from_df_skips_missing_url_and_strips_trailing_slash() -> None:
+ from website_profiling.db.crawl_store import _crawl_rows_from_df
+
+ df = pd.DataFrame(
+ [
+ {"url": "https://a.com/", "status": 200, "title": "A", "x": 1},
+ {"url": "", "status": 200, "title": "B", "x": 2},
+ ]
+ )
+ rows = _crawl_rows_from_df(df, crawl_run_id=7)
+ assert rows[0][0] == 7
+ assert rows[0][1] == "https://a.com"
+ assert len(rows) == 1
+
+
+def test_read_report_payload_handles_missing_and_dict() -> None:
+ from website_profiling.db.report_store import read_report_payload
+
+ conn = FakeConn()
+ conn.set_next_cursor(FakeCursor(fetchone_value=None))
+ assert read_report_payload(conn) is None # type: ignore[arg-type]
+
+ conn = FakeConn()
+ conn.set_next_cursor(FakeCursor(fetchone_value={"data": {"ok": True}}))
+ assert read_report_payload(conn) == {"ok": True} # type: ignore[arg-type]
+
+
+def test_llm_cache_roundtrip_parsing() -> None:
+ from website_profiling.db.llm_cache_store import read_llm_cache, read_llm_cache_batch
+
+ conn = FakeConn()
+ conn.set_next_cursor(FakeCursor(fetchone_value={"response_json": {"a": 1}}))
+ assert read_llm_cache(conn, "k") == '{"a": 1}' # type: ignore[arg-type]
+
+ conn = FakeConn()
+ conn.set_next_cursor(
+ FakeCursor(
+ fetchall_value=[
+ {"cache_key": "k1", "response_json": {"x": 1}},
+ {"cache_key": "k2", "response_json": '{"y": 2}'},
+ {"cache_key": "k3", "response_json": "not-json"},
+ ]
+ )
+ )
+ out = read_llm_cache_batch(conn, ["k1", "k2", "k3"]) # type: ignore[arg-type]
+ assert out["k1"] == {"x": 1}
+ assert out["k2"] == {"y": 2}
+ assert "k3" not in out
+
+
+def test_edges_and_nodes_write_and_read_paths() -> None:
+ from website_profiling.db.crawl_store import read_edges, read_nodes, write_edges, write_nodes
+
+ conn = FakeConn()
+ # read_edges: return two rows
+ conn.set_next_cursor(FakeCursor(fetchall_value=[{"from_url": "a", "to_url": "b"}, {"from_url": "b", "to_url": "c"}]))
+ assert read_edges(conn, run_id=1) == [("a", "b"), ("b", "c")] # type: ignore[arg-type]
+
+ # write_edges: with explicit run id, should delete then insert + commit
+ conn = FakeConn()
+ write_edges(conn, [("https://a.com/", "https://b.com/")], crawl_run_id=5) # type: ignore[arg-type]
+ assert conn.commits == 1
+ assert any("DELETE FROM edges" in s for (s, _p) in conn.executed)
+
+ # write_nodes: empty df clears table
+ conn = FakeConn()
+ write_nodes(conn, pd.DataFrame(), crawl_run_id=2) # type: ignore[arg-type]
+ assert conn.commits == 1
+
+ # read_nodes: rows -> dataframe with columns
+ conn = FakeConn()
+ conn.set_next_cursor(FakeCursor(fetchall_value=[{"url": "u1", "count": 2}, {"url": "u2", "count": 1}]))
+ df = read_nodes(conn, run_id=3) # type: ignore[arg-type]
+ assert df.shape[0] == 2
+
+
+def test_lighthouse_store_summary_and_run_id() -> None:
+ from website_profiling.db.lighthouse_store import read_lighthouse_summary, write_lighthouse_run, write_lighthouse_summary
+
+ conn = FakeConn()
+ write_lighthouse_summary(conn, {"a": 1}) # type: ignore[arg-type]
+ assert conn.commits == 1
+
+ conn = FakeConn()
+ conn.set_next_cursor(FakeCursor(fetchone_value={"data": {"ok": True}}))
+ assert read_lighthouse_summary(conn) == {"ok": True} # type: ignore[arg-type]
+
+ conn = FakeConn()
+ conn.set_next_cursor(FakeCursor(fetchone_value={"id": 9}))
+ rid = write_lighthouse_run(conn, url="u", strategy="mobile", run_index=0, data={"x": 1}) # type: ignore[arg-type]
+ assert rid == 9
+
diff --git a/tests/test_final_push80.py b/tests/test_final_push80.py
new file mode 100644
index 00000000..ecc6ce49
--- /dev/null
+++ b/tests/test_final_push80.py
@@ -0,0 +1,267 @@
+from __future__ import annotations
+
+import argparse
+import types
+
+import pandas as pd
+import pytest
+
+
+class C:
+ def __init__(self, rows=None, row=None):
+ self._rows = rows or []
+ self._row = row
+ self.executed = []
+ self.commits = 0
+
+ def execute(self, sql, params=None):
+ self.executed.append((sql, params))
+ if "SELECT data FROM lighthouse_runs ORDER BY id DESC LIMIT 1" in sql:
+ return C(row={"data": {"latest": 1}})
+ if "RETURNING id" in sql:
+ return C(row={"id": 7})
+ if "ORDER BY id DESC LIMIT 1" in sql:
+ return C(row={"id": 2})
+ if "SELECT url, data FROM crawl_results" in sql:
+ return C(rows=[{"url": "u", "data": {"viewport_present": True, "noindex": False, "has_schema": True}}])
+ if "SELECT from_url, to_url FROM edges" in sql:
+ return C(rows=[{"from_url": "a", "to_url": "b"}])
+ if "SELECT url, count FROM nodes" in sql:
+ return C(rows=[{"url": "u", "count": 1}])
+ if "SELECT data FROM lighthouse_runs WHERE id" in sql:
+ return C(row={"data": {"x": 1}})
+ if "SELECT data FROM lighthouse_runs ORDER BY id DESC LIMIT 1" in sql:
+ return C(row={"data": {"latest": 1}})
+ return C()
+
+ def fetchone(self):
+ return self._row
+
+ def fetchall(self):
+ return list(self._rows)
+
+ def commit(self):
+ self.commits += 1
+
+ def transaction(self):
+ class CM:
+ def __enter__(self_non):
+ return None
+
+ def __exit__(self_non, _t, _v, _tb):
+ return False
+
+ return CM()
+
+ def cursor(self):
+ cur = C()
+
+ class CM:
+ def __enter__(self_non):
+ return cur
+
+ def __exit__(self_non, _t, _v, _tb):
+ return False
+
+ return CM()
+
+ def executemany(self, _sql, _params):
+ return None
+
+
+def test_crawl_store_write_read_paths(monkeypatch):
+ from website_profiling.db import crawl_store as cs
+
+ conn = C()
+ # write_crawl with explicit run id
+ df = pd.DataFrame([{"url": "https://a.com/", "status": 200, "title": "A"}])
+ cs.write_crawl(conn, df, crawl_run_id=5) # type: ignore[arg-type]
+ assert any("DELETE FROM crawl_results WHERE crawl_run_id" in s for s, _ in conn.executed)
+
+ # write_crawl when run id missing should create crawl_run
+ conn2 = C()
+ monkeypatch.setattr(cs, "get_latest_crawl_run_id", lambda _c: None)
+ cs.write_crawl(conn2, df, crawl_run_id=None) # type: ignore[arg-type]
+ assert any("INSERT INTO crawl_runs" in s for s, _ in conn2.executed)
+
+ # read_crawl bool coercion
+ out = cs.read_crawl(C(), run_id=1) # type: ignore[arg-type]
+ assert bool(out.loc[0, "viewport_present"]) is True
+
+ assert cs.read_edges(C(), run_id=1) == [("a", "b")] # type: ignore[arg-type]
+ assert list(cs.read_nodes(C(), run_id=1).columns) == ["url", "count"] # type: ignore[arg-type]
+
+
+def test_lighthouse_store_run_json_and_latest_and_page_summary():
+ from website_profiling.db import lighthouse_store as ls
+
+ conn = C()
+ assert ls.read_lighthouse_run_json(conn, 1) == {"x": 1} # type: ignore[arg-type]
+ assert ls.read_latest_lighthouse_run_json(conn) == {"latest": 1} # type: ignore[arg-type]
+ ls.write_lighthouse_page_summary(conn, "https://a.com", {"s": 1}) # type: ignore[arg-type]
+ assert conn.commits >= 1
+
+
+def test_google_run_runtimeerror_branch(monkeypatch):
+ from website_profiling.commands import google_cmd
+
+ class Ctx:
+ def __enter__(self):
+ return object()
+
+ def __exit__(self, _t, _v, _tb):
+ return False
+
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.auth",
+ types.SimpleNamespace(build_credentials=lambda *_a, **_k: None, read_secrets=lambda *_a, **_k: {}),
+ )
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.fetch",
+ types.SimpleNamespace(fetch_google_data=lambda **_k: (_ for _ in ()).throw(RuntimeError("bad")), list_properties=lambda *_a, **_k: {}),
+ )
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.store",
+ types.SimpleNamespace(write_google_data=lambda *_a, **_k: None),
+ )
+
+ import sys as _sys
+ import types as _types
+
+ g = _types.ModuleType("google")
+ ga = _types.ModuleType("google.auth")
+ ge = _types.ModuleType("google.auth.exceptions")
+ ge.RefreshError = ValueError
+ ga.exceptions = ge
+ g.auth = ga
+ monkeypatch.setitem(_sys.modules, "google", g)
+ monkeypatch.setitem(_sys.modules, "google.auth", ga)
+ monkeypatch.setitem(_sys.modules, "google.auth.exceptions", ge)
+
+ import website_profiling.db as db
+
+ monkeypatch.setattr(db, "db_session", lambda: Ctx())
+ monkeypatch.setattr(db, "get_latest_crawl_run_id", lambda _c: None)
+ monkeypatch.setattr(db, "read_crawl", lambda *_a, **_k: pd.DataFrame())
+
+ with pytest.raises(SystemExit) as e:
+ google_cmd.run({"start_url": "https://a.com"}, "/tmp", lambda _k, d: d, argparse.Namespace(list_properties=False, test=False))
+ assert e.value.code == 1
+
+
+def test_google_run_google_test_success_branch(monkeypatch):
+ from website_profiling.commands import google_cmd
+
+ import sys as _sys
+ import types as _types
+
+ g = _types.ModuleType("google")
+ ga = _types.ModuleType("google.auth")
+ ge = _types.ModuleType("google.auth.exceptions")
+ ge.RefreshError = RuntimeError
+ ga.exceptions = ge
+ g.auth = ga
+ monkeypatch.setitem(_sys.modules, "google", g)
+ monkeypatch.setitem(_sys.modules, "google.auth", ga)
+ monkeypatch.setitem(_sys.modules, "google.auth.exceptions", ge)
+
+ monkeypatch.setitem(
+ _sys.modules,
+ "website_profiling.integrations.google.auth",
+ _types.SimpleNamespace(
+ build_credentials=lambda *_a, **_k: object(),
+ read_secrets=lambda *_a, **_k: {"gscSiteUrl": "https://prop/", "ga4PropertyId": "123"},
+ ),
+ )
+ monkeypatch.setitem(
+ _sys.modules,
+ "website_profiling.integrations.google.gsc",
+ _types.SimpleNamespace(
+ list_gsc_sites=lambda _c: ["https://prop/"],
+ resolve_gsc_site_url=lambda s, _sites: (s, None),
+ probe_gsc_site=lambda *_a, **_k: (True, "ok"),
+ describe_gsc_site_mismatch=lambda *_a, **_k: "mismatch",
+ ),
+ )
+ monkeypatch.setitem(
+ _sys.modules,
+ "website_profiling.integrations.google.ga4",
+ _types.SimpleNamespace(
+ list_ga4_properties=lambda _c: ([{"id": "123", "displayName": "Main"}], None),
+ probe_ga4_property=lambda *_a, **_k: (True, "ok"),
+ ),
+ )
+
+ with pytest.raises(SystemExit) as e:
+ google_cmd._run_google_test(None)
+ assert e.value.code == 0
+
+
+def test_crawl_store_more_branches(monkeypatch):
+ from website_profiling.db import crawl_store as cs
+
+ row = pd.Series({"url": "https://a.com", "status": 200, "x": 1})
+ out = cs._df_row_to_crawl_json(row)
+ assert out["status"] == 200
+ assert "url" not in out
+
+ monkeypatch.setattr(cs, "get_crawl_run_info", lambda _c, _rid: None)
+ assert cs._canonical_domain_from_report(C(), {"top_pages": [{"url": "https://X.com/p"}]}) == "x.com" # type: ignore[arg-type]
+
+ conn = C()
+ cs.write_edges(conn, [("https://a.com/", "https://b.com/")], crawl_run_id=None) # type: ignore[arg-type]
+ assert conn.commits >= 1
+
+ monkeypatch.setattr(cs, "get_latest_crawl_run_id", lambda _c: None)
+ assert cs.read_edges(conn, run_id=None) == [] # type: ignore[arg-type]
+
+
+def test_google_run_google_test_error_branches(monkeypatch):
+ from website_profiling.commands import google_cmd
+ import sys as _sys
+ import types as _types
+
+ g = _types.ModuleType("google")
+ ga = _types.ModuleType("google.auth")
+ ge = _types.ModuleType("google.auth.exceptions")
+ ge.RefreshError = RuntimeError
+ ga.exceptions = ge
+ g.auth = ga
+ monkeypatch.setitem(_sys.modules, "google", g)
+ monkeypatch.setitem(_sys.modules, "google.auth", ga)
+ monkeypatch.setitem(_sys.modules, "google.auth.exceptions", ge)
+
+ # Branch: resolve mismatch and GA4 probe failure -> exit 1
+ monkeypatch.setitem(
+ _sys.modules,
+ "website_profiling.integrations.google.auth",
+ _types.SimpleNamespace(
+ build_credentials=lambda *_a, **_k: object(),
+ read_secrets=lambda *_a, **_k: {"gscSiteUrl": "https://bad/", "ga4PropertyId": "999"},
+ ),
+ )
+ monkeypatch.setitem(
+ _sys.modules,
+ "website_profiling.integrations.google.gsc",
+ _types.SimpleNamespace(
+ list_gsc_sites=lambda _c: ["https://good/"],
+ resolve_gsc_site_url=lambda *_a, **_k: (None, "bad site"),
+ probe_gsc_site=lambda *_a, **_k: (False, "no"),
+ describe_gsc_site_mismatch=lambda *_a, **_k: "mismatch",
+ ),
+ )
+ monkeypatch.setitem(
+ _sys.modules,
+ "website_profiling.integrations.google.ga4",
+ _types.SimpleNamespace(
+ list_ga4_properties=lambda _c: ([], "list err"),
+ probe_ga4_property=lambda *_a, **_k: (False, "ga4 bad"),
+ ),
+ )
+ with pytest.raises(SystemExit) as e:
+ google_cmd._run_google_test(None)
+ assert e.value.code == 1
+
diff --git a/tests/test_google_cmd_more.py b/tests/test_google_cmd_more.py
new file mode 100644
index 00000000..757dce63
--- /dev/null
+++ b/tests/test_google_cmd_more.py
@@ -0,0 +1,54 @@
+import argparse
+import types
+
+import pytest
+
+
+def test_google_cmd_test_flag_calls_test(monkeypatch) -> None:
+ from website_profiling.commands import google_cmd
+
+ called = {"n": 0}
+
+ def fake_run_test(_p):
+ called["n"] += 1
+
+ monkeypatch.setattr(google_cmd, "_run_google_test", fake_run_test)
+ # Ensure imports inside run() don't blow up
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.auth",
+ types.SimpleNamespace(build_credentials=lambda *_a, **_k: None, read_secrets=lambda *_a, **_k: {}),
+ )
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.fetch",
+ types.SimpleNamespace(fetch_google_data=lambda *_a, **_k: {}, list_properties=lambda *_a, **_k: {}),
+ )
+
+ args = argparse.Namespace(list_properties=False, test=True)
+ google_cmd.run({"google_credentials_path": "rel.json"}, cwd="/cwd", path=lambda k, d: d, args=args)
+ assert called["n"] == 1
+
+
+def test_google_cmd_list_properties_error_exits_1(monkeypatch) -> None:
+ from website_profiling.commands import google_cmd
+
+ def boom(_p=None):
+ raise RuntimeError("nope")
+
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.fetch",
+ types.SimpleNamespace(list_properties=boom, fetch_google_data=None),
+ )
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.auth",
+ types.SimpleNamespace(build_credentials=lambda *_a, **_k: None, read_secrets=lambda *_a, **_k: {}),
+ )
+
+ args = argparse.Namespace(list_properties=True, test=False)
+ with pytest.raises(SystemExit) as e:
+ google_cmd.run({"google_credentials_path": ""}, cwd="/tmp", path=lambda k, d: d, args=args)
+ assert e.value.code == 1
+
diff --git a/tests/test_high_coverage_push2.py b/tests/test_high_coverage_push2.py
new file mode 100644
index 00000000..3bd71ff3
--- /dev/null
+++ b/tests/test_high_coverage_push2.py
@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import argparse
+import types
+
+import pandas as pd
+
+
+def test_pipeline_run_branch_selection(monkeypatch):
+ from website_profiling.commands import pipeline_cmd
+
+ calls = {"crawl": 0, "lh_pages": 0, "lh_one": 0, "report": 0, "plot": 0}
+ monkeypatch.setattr(pipeline_cmd, "_run_crawl", lambda *_a, **_k: calls.__setitem__("crawl", calls["crawl"] + 1))
+ monkeypatch.setattr(pipeline_cmd, "_run_lighthouse_on_pages", lambda *_a, **_k: calls.__setitem__("lh_pages", calls["lh_pages"] + 1))
+ monkeypatch.setattr(pipeline_cmd, "_run_single_lighthouse", lambda *_a, **_k: calls.__setitem__("lh_one", calls["lh_one"] + 1))
+ monkeypatch.setattr(pipeline_cmd, "_run_report", lambda *_a, **_k: calls.__setitem__("report", calls["report"] + 1))
+ monkeypatch.setattr(pipeline_cmd, "_run_plot", lambda *_a, **_k: calls.__setitem__("plot", calls["plot"] + 1))
+
+ cfg = {
+ "run_crawl": "true",
+ "run_lighthouse_on_pages": "true",
+ "run_lighthouse": "false",
+ "run_report": "true",
+ "run_plot": "true",
+ }
+ pipeline_cmd.run(cfg, argparse.Namespace(command=None))
+ assert calls == {"crawl": 1, "lh_pages": 1, "lh_one": 0, "report": 1, "plot": 1}
+
+
+def test_pipeline_run_single_lighthouse_when_enabled_in_config(monkeypatch):
+ from website_profiling.commands import pipeline_cmd
+
+ called = {"lh": 0}
+ monkeypatch.setattr(pipeline_cmd, "_run_single_lighthouse", lambda *_a, **_k: called.__setitem__("lh", called["lh"] + 1))
+ monkeypatch.setattr(pipeline_cmd, "_run_crawl", lambda *_a, **_k: None)
+ monkeypatch.setattr(pipeline_cmd, "_run_report", lambda *_a, **_k: None)
+ monkeypatch.setattr(pipeline_cmd, "_run_plot", lambda *_a, **_k: None)
+ monkeypatch.setattr(pipeline_cmd, "_run_lighthouse_on_pages", lambda *_a, **_k: None)
+ pipeline_cmd.run({"run_lighthouse": "true"}, argparse.Namespace(command=None))
+ assert called["lh"] == 1
+
+
+def test_run_single_lighthouse_exits_on_nonzero(monkeypatch):
+ from website_profiling.commands import pipeline_cmd
+
+ monkeypatch.setattr(pipeline_cmd, "require_lighthouse_url", lambda _cfg: "https://a.com")
+ monkeypatch.setattr(pipeline_cmd, "lighthouse_work_dir", lambda: "/tmp/w")
+ monkeypatch.setattr(pipeline_cmd, "cleanup_lighthouse_work_dir", lambda _p: None)
+ monkeypatch.setitem(__import__("sys").modules, "website_profiling.lighthouse.runner", types.SimpleNamespace(main=lambda **_k: 2))
+ import pytest
+
+ with pytest.raises(SystemExit) as e:
+ pipeline_cmd._run_single_lighthouse({}, True)
+ assert e.value.code == 2
+
+
+def test_run_report_and_plot_paths(monkeypatch):
+ from website_profiling.commands import pipeline_cmd
+
+ monkeypatch.setattr(pipeline_cmd, "require_start_url", lambda *_a, **_k: "https://a.com")
+ monkeypatch.setattr(pipeline_cmd, "should_enrich_keywords_after_report", lambda _cfg: False)
+ monkeypatch.setitem(__import__("sys").modules, "website_profiling.reporting.builder", types.SimpleNamespace(run_simple_report=lambda **_k: "report.json"))
+ monkeypatch.setitem(__import__("sys").modules, "website_profiling.tools.plot", types.SimpleNamespace(run_plot=lambda **_k: 0))
+ pipeline_cmd._run_report({}, True)
+ pipeline_cmd._run_plot({}, True)
+
+
+def test_parse_tech_stack_and_wappalyzer_fallbacks(monkeypatch):
+ from bs4 import BeautifulSoup
+ from website_profiling.common import detect_tech_wappalyzer, parse_tech_stack
+
+ html = '__NEXT_DATA__ jquery bootstrap'
+ soup = BeautifulSoup(html, "lxml")
+ stack = parse_tech_stack(soup, {"Server": "nginx", "cf-ray": "x"}, "https://a.com")
+ assert "Next.js" in stack or "Drupal" in stack
+
+ class FakeW:
+ def analyze_with_versions_and_categories(self, _web):
+ return {"WordPress": {"versions": [], "categories": ["CMS"]}}
+
+ out = detect_tech_wappalyzer("https://a.com", html, {}, soup, FakeW())
+ # Depending on Wappalyzer/WebPage availability this may fall back to parse_tech_stack.
+ assert out.startswith("[") and out.endswith("]")
+
+
+def test_pool_lifecycle_and_db_session(monkeypatch):
+ from website_profiling.db import pool
+
+ class FakeConnCtx:
+ def __enter__(self):
+ return object()
+
+ def __exit__(self, _t, _v, _tb):
+ return False
+
+ class FakePool:
+ def __init__(self, **_kwargs):
+ self.closed = False
+
+ def connection(self, timeout=5):
+ return FakeConnCtx()
+
+ def close(self):
+ self.closed = True
+
+ monkeypatch.setattr(pool, "ConnectionPool", FakePool)
+ monkeypatch.setattr(pool, "get_database_url", lambda: "postgres://u:p@h/db")
+ pool.close_db_pool()
+ with pool.db_session() as conn:
+ assert conn is not None
+ pool.close_db_pool()
+
diff --git a/tests/test_llm_config.py b/tests/test_llm_config.py
new file mode 100644
index 00000000..78ec8338
--- /dev/null
+++ b/tests/test_llm_config.py
@@ -0,0 +1,37 @@
+"""Tests for LLM config loading (PostgreSQL)."""
+from __future__ import annotations
+
+import os
+
+import pytest
+
+from website_profiling.db import db_session
+from website_profiling.db.storage import write_llm_config
+from website_profiling.llm_config import llm_is_enabled, load_llm_config_from_db
+
+
+@pytest.fixture(scope="module")
+def require_database_url():
+ if not (os.environ.get("DATABASE_URL") or "").strip():
+ pytest.skip("DATABASE_URL not set — start Postgres and run alembic upgrade head")
+
+
+def test_load_llm_config_from_db(require_database_url):
+ with db_session() as conn:
+ write_llm_config(
+ conn,
+ {
+ "llm_enabled": "true",
+ "llm_provider": "ollama",
+ "llm_model": "llama3.2",
+ },
+ secret_keys=set(),
+ )
+ cfg = load_llm_config_from_db()
+ assert cfg.get("llm_provider") == "ollama"
+ assert llm_is_enabled(cfg)
+
+
+def test_llm_disabled_by_default():
+ assert not llm_is_enabled({})
+ assert not llm_is_enabled({"llm_enabled": "false", "llm_provider": "openai"})
diff --git a/tests/test_llm_parse.py b/tests/test_llm_parse.py
new file mode 100644
index 00000000..c274d89a
--- /dev/null
+++ b/tests/test_llm_parse.py
@@ -0,0 +1,14 @@
+"""Tests for LLM JSON parsing."""
+from __future__ import annotations
+
+from website_profiling.llm.base import parse_json_response
+
+
+def test_parse_json_response_plain():
+ assert parse_json_response('{"pages": []}') == {"pages": []}
+
+
+def test_parse_json_response_markdown_fence():
+ text = 'Here is JSON:\n{"clusters": [{"top_keyword": "seo"}]}\n'
+ out = parse_json_response(text)
+ assert "clusters" in out
diff --git a/tests/test_load_config_from_db.py b/tests/test_load_config_from_db.py
new file mode 100644
index 00000000..ed502ce4
--- /dev/null
+++ b/tests/test_load_config_from_db.py
@@ -0,0 +1,28 @@
+"""load_config_from_db error handling."""
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, patch
+
+from website_profiling.config import load_config_from_db
+
+
+def test_returns_empty_when_database_url_unset(monkeypatch, capsys):
+ monkeypatch.delenv("DATABASE_URL", raising=False)
+ assert load_config_from_db() == {}
+ assert capsys.readouterr().err == ""
+
+
+def test_logs_warning_on_db_error(monkeypatch, capsys):
+ monkeypatch.setenv("DATABASE_URL", "postgres://u:p@127.0.0.1:5432/test")
+
+ mock_session = MagicMock()
+ mock_session.__enter__.side_effect = RuntimeError("connection refused")
+
+ with patch("website_profiling.db.db_session", return_value=mock_session):
+ result = load_config_from_db()
+
+ assert result == {}
+ err = capsys.readouterr().err
+ assert "Could not load pipeline_config" in err
+ assert "connection refused" in err
diff --git a/tests/test_missing_points_batch.py b/tests/test_missing_points_batch.py
new file mode 100644
index 00000000..a5d3020e
--- /dev/null
+++ b/tests/test_missing_points_batch.py
@@ -0,0 +1,159 @@
+from __future__ import annotations
+
+import argparse
+import types
+
+import pandas as pd
+import pytest
+
+
+def test_cfg_int_uses_legacy_key() -> None:
+ from website_profiling.analysis.local import _cfg_int
+
+ cfg = {"ml_fuzzy_threshold": "95"}
+ assert _cfg_int(cfg, "analysis_fuzzy_threshold", 92) == 95
+
+
+def test_compute_duplicate_groups_disabled_returns_empty() -> None:
+ from website_profiling.analysis.local import compute_duplicate_groups
+
+ df = pd.DataFrame([{"url": "https://a.com", "status": "200", "content_type": "text/html"}])
+ groups, mapping = compute_duplicate_groups(df, {"enable_duplicate_detection": "false"})
+ assert groups == []
+ assert mapping == {}
+
+
+def test_compute_duplicate_groups_basic_cluster(monkeypatch) -> None:
+ from website_profiling.analysis import local
+
+ monkeypatch.setattr(local, "_import_rapidfuzz", lambda: types.SimpleNamespace(token_set_ratio=lambda a, b: 100))
+ # Avoid relying on normalize_fingerprint_text internals for this unit
+ monkeypatch.setattr(local, "normalize_fingerprint_text", lambda _row: "this is enough textual content for duplicate check")
+ df = pd.DataFrame(
+ [
+ {"url": "https://a.com/1", "status": "200", "content_type": "text/html"},
+ {"url": "https://a.com/2", "status": "200", "content_type": "text/html"},
+ ]
+ )
+ groups, mapping = local.compute_duplicate_groups(df, {"enable_duplicate_detection": "true"})
+ assert len(groups) == 1
+ assert mapping["https://a.com/1"].startswith("dup_")
+
+
+def test_compute_language_signals_disabled() -> None:
+ from website_profiling.analysis.local import compute_language_signals
+
+ df = pd.DataFrame([{"url": "https://a.com", "status": "200"}])
+ by_url, summary = compute_language_signals(df, {"enable_language_detection": "false"})
+ assert by_url == {}
+ assert summary["mixed_site"] is False
+
+
+def test_run_local_enrichment_collects_import_errors(monkeypatch) -> None:
+ from website_profiling.analysis import local
+
+ monkeypatch.setattr(local, "compute_duplicate_groups", lambda *_a, **_k: (_ for _ in ()).throw(ImportError("dup import missing")))
+ monkeypatch.setattr(local, "compute_language_signals", lambda *_a, **_k: (_ for _ in ()).throw(ImportError("lang import missing")))
+ out = local.run_local_enrichment(pd.DataFrame([{"url": "x"}]), {"enable_duplicate_detection": "true"})
+ assert len(out["ml_errors"]) == 2
+
+
+def test_merge_bundles_merges_map_like_fields() -> None:
+ from website_profiling.analysis.local import merge_bundles
+
+ local = {"language_by_url": {"a": "en"}, "ml_errors": ["e1"]}
+ llm = {"language_by_url": {"b": "fr"}, "ml_errors": ["e2"]}
+ out = merge_bundles(local, llm)
+ assert out["language_by_url"] == {"a": "en", "b": "fr"}
+ assert out["ml_errors"] == ["e1", "e2"]
+
+
+def test_merge_analysis_into_payload_updates_links() -> None:
+ from website_profiling.analysis.local import merge_analysis_into_payload
+
+ payload = {
+ "links": [
+ {"url": "https://a.com/x", "page_analysis": {"signals": {"language": "old"}}},
+ ]
+ }
+ bundle = {
+ "content_duplicates": [{"id": "dup_0"}],
+ "url_duplicate_group_id": {"https://a.com/x": "dup_0"},
+ "language_by_url": {"https://a.com/x": "en"},
+ "spacy_by_url": {"https://a.com/x": [{"text": "Paris"}]},
+ "keyphrases_by_url": {"https://a.com/x": ["k1"]},
+ "language_summary": {"mixed_site": False},
+ }
+ merge_analysis_into_payload(payload, bundle)
+ rec = payload["links"][0]
+ assert rec["duplicate_group_id"] == "dup_0"
+ assert rec["detected_language"] == "en"
+ assert rec["page_analysis"]["signals"]["language"] == "en"
+
+
+def test_pool_env_int_and_data_dir(monkeypatch) -> None:
+ from website_profiling.db.pool import _env_int, get_data_dir
+
+ monkeypatch.setenv("X_NUM", "0")
+ assert _env_int("X_NUM", 7) == 1
+ monkeypatch.setenv("X_NUM", "bad")
+ assert _env_int("X_NUM", 7) == 7
+ monkeypatch.setenv("DATA_DIR", "/tmp/abc")
+ assert get_data_dir() == "/tmp/abc"
+
+
+def test_build_parser_accepts_known_command() -> None:
+ from website_profiling.commands.config_resolve import build_parser
+
+ parser = build_parser()
+ args = parser.parse_args(["google"])
+ assert args.command == "google"
+
+
+def test_google_cmd_run_fetch_success(monkeypatch) -> None:
+ from website_profiling.commands import google_cmd
+
+ class _Ctx:
+ def __enter__(self):
+ return object()
+
+ def __exit__(self, _t, _v, _tb):
+ return False
+
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.auth",
+ types.SimpleNamespace(build_credentials=lambda *_a, **_k: None, read_secrets=lambda *_a, **_k: {}),
+ )
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.fetch",
+ types.SimpleNamespace(fetch_google_data=lambda **_k: {"errors": []}, list_properties=lambda *_a, **_k: {}),
+ )
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.store",
+ types.SimpleNamespace(write_google_data=lambda *_a, **_k: None),
+ )
+ import types as _types
+ import sys as _sys
+
+ google_mod = _types.ModuleType("google")
+ auth_mod = _types.ModuleType("google.auth")
+ exc_mod = _types.ModuleType("google.auth.exceptions")
+ exc_mod.RefreshError = RuntimeError
+ auth_mod.exceptions = exc_mod
+ google_mod.auth = auth_mod
+ monkeypatch.setitem(_sys.modules, "google", google_mod)
+ monkeypatch.setitem(_sys.modules, "google.auth", auth_mod)
+ monkeypatch.setitem(_sys.modules, "google.auth.exceptions", exc_mod)
+ import website_profiling.db as db
+
+ monkeypatch.setattr(db, "db_session", lambda: _Ctx())
+ monkeypatch.setattr(db, "get_latest_crawl_run_id", lambda _c: None)
+ monkeypatch.setattr(db, "read_crawl", lambda _c, _rid: pd.DataFrame())
+
+ with pytest.raises(SystemExit) as e:
+ google_cmd.run({"start_url": "https://a.com"}, cwd="/tmp", path=lambda _k, d: d, args=argparse.Namespace(list_properties=False, test=False))
+ assert e.value.code == 0
+
diff --git a/tests/test_more_coverage_push.py b/tests/test_more_coverage_push.py
new file mode 100644
index 00000000..d619cda3
--- /dev/null
+++ b/tests/test_more_coverage_push.py
@@ -0,0 +1,416 @@
+from __future__ import annotations
+
+import argparse
+import types
+
+import pandas as pd
+import pytest
+
+
+class _Cursor:
+ def __init__(self, rows=None, row=None):
+ self._rows = rows or []
+ self._row = row
+ self.executed = []
+
+ def execute(self, sql, params=None):
+ self.executed.append((sql, params))
+ return self
+
+ def fetchall(self):
+ return list(self._rows)
+
+ def fetchone(self):
+ return self._row
+
+
+class _Conn:
+ def __init__(self, by_sql=None):
+ self.by_sql = by_sql or {}
+ self.executed = []
+ self.commits = 0
+
+ def execute(self, sql, params=None):
+ self.executed.append((sql, params))
+ for key, cur in self.by_sql.items():
+ if key in sql:
+ return cur
+ return _Cursor()
+
+ def commit(self):
+ self.commits += 1
+
+ def cursor(self):
+ cur = _Cursor()
+
+ class _CM:
+ def __enter__(self_non):
+ return cur
+
+ def __exit__(self_non, _t, _v, _tb):
+ return False
+
+ return _CM()
+
+ def transaction(self):
+ class _CM:
+ def __enter__(self_non):
+ return None
+
+ def __exit__(self_non, _t, _v, _tb):
+ return False
+
+ return _CM()
+
+
+def test_keywords_expand_only_no_seeds_exits_1(capsys):
+ from website_profiling.commands import keywords_cmd
+
+ with pytest.raises(SystemExit) as e:
+ keywords_cmd.run({"keyword_seeds": ""}, argparse.Namespace(expand_only=True, enrich_google=False))
+ assert e.value.code == 1
+ assert "No keyword_seeds configured" in capsys.readouterr().out
+
+
+def test_keywords_enrich_google_success(monkeypatch):
+ from website_profiling.commands import keywords_cmd
+
+ called = {"n": 0}
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.keyword_enrich",
+ types.SimpleNamespace(run_enrichment=lambda _cfg: called.__setitem__("n", called["n"] + 1)),
+ )
+ with pytest.raises(SystemExit) as e:
+ keywords_cmd.run({}, argparse.Namespace(expand_only=False, enrich_google=True))
+ assert e.value.code == 0
+ assert called["n"] == 1
+
+
+def test_keywords_enrich_google_failure(monkeypatch):
+ from website_profiling.commands import keywords_cmd
+
+ def _boom(_cfg):
+ raise RuntimeError("x")
+
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.keyword_enrich",
+ types.SimpleNamespace(run_enrichment=_boom),
+ )
+ with pytest.raises(SystemExit) as e:
+ keywords_cmd.run({}, argparse.Namespace(expand_only=False, enrich_google=True))
+ assert e.value.code == 1
+
+
+def test_keywords_main_flow_nonzero_exit(monkeypatch):
+ from website_profiling.commands import keywords_cmd
+
+ monkeypatch.setattr(keywords_cmd, "require_start_url", lambda *_a, **_k: "https://a.com")
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.tools.keywords",
+ types.SimpleNamespace(main=lambda **_k: 3),
+ )
+ with pytest.raises(SystemExit) as e:
+ keywords_cmd.run({"enable_google_suggest": "false"}, argparse.Namespace(expand_only=False, enrich_google=False))
+ assert e.value.code == 3
+
+
+def test_keywords_main_flow_enrichment_triggered(monkeypatch):
+ from website_profiling.commands import keywords_cmd
+
+ called = {"enrich": 0}
+ monkeypatch.setattr(keywords_cmd, "require_start_url", lambda *_a, **_k: "https://a.com")
+ monkeypatch.setattr(keywords_cmd, "google_db_has_gsc", lambda: True)
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.tools.keywords",
+ types.SimpleNamespace(main=lambda **_k: 0),
+ )
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.integrations.google.keyword_enrich",
+ types.SimpleNamespace(run_enrichment=lambda _cfg: called.__setitem__("enrich", called["enrich"] + 1)),
+ )
+ with pytest.raises(SystemExit) as e:
+ keywords_cmd.run({"enable_google_suggest": "false"}, argparse.Namespace(expand_only=False, enrich_google=False))
+ assert e.value.code == 0
+ assert called["enrich"] == 1
+
+
+def test_google_db_has_gsc_true_and_false(monkeypatch):
+ from website_profiling.commands import config_resolve
+
+ class _Ctx:
+ def __init__(self, conn):
+ self.conn = conn
+
+ def __enter__(self):
+ return self.conn
+
+ def __exit__(self, _t, _v, _tb):
+ return False
+
+ conn_true = _Conn({"SELECT data FROM google_data": _Cursor(row={"data": {"gsc_full": {"top_queries": [1]}}})})
+ monkeypatch.setitem(__import__("sys").modules, "website_profiling.db", types.SimpleNamespace(db_session=lambda: _Ctx(conn_true)))
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.db.storage",
+ types.SimpleNamespace(_parse_json_field=lambda v: v),
+ )
+ assert config_resolve.google_db_has_gsc() is True
+
+ conn_false = _Conn({"SELECT data FROM google_data": _Cursor(row={"data": {"gsc_full": {}}})})
+ monkeypatch.setitem(__import__("sys").modules, "website_profiling.db", types.SimpleNamespace(db_session=lambda: _Ctx(conn_false)))
+ assert config_resolve.google_db_has_gsc() is False
+
+
+def test_historical_backup_skip_ci_and_timeout(monkeypatch, tmp_path):
+ from website_profiling.db import historical
+
+ monkeypatch.setenv("CI", "true")
+ assert historical.backup_db_if_exists(skip_in_ci=True) is None
+ monkeypatch.setenv("CI", "false")
+ monkeypatch.setattr(historical, "get_data_dir", lambda: str(tmp_path))
+ monkeypatch.setattr(historical, "get_database_url", lambda: "postgres://u:p@h/db")
+ import subprocess as _sp
+
+ monkeypatch.setattr(
+ historical.subprocess,
+ "run",
+ lambda *a, **k: (_ for _ in ()).throw(_sp.TimeoutExpired(cmd="pg_dump", timeout=1)),
+ )
+ assert historical.backup_db_if_exists(skip_in_ci=False) is None
+
+
+def test_ensure_crawl_tables_cleared_commits():
+ from website_profiling.db.historical import ensure_crawl_tables_cleared
+
+ conn = _Conn()
+ ensure_crawl_tables_cleared(conn) # type: ignore[arg-type]
+ assert conn.commits == 1
+ assert any("TRUNCATE crawl_results, edges, nodes" in s for s, _ in conn.executed)
+
+
+def test_read_lh_runs_by_url_and_page_summaries():
+ from website_profiling.db.lighthouse_store import read_lh_runs_by_url, read_lighthouse_page_summaries
+
+ conn = _Conn(
+ {
+ "SELECT id, url FROM lighthouse_runs": _Cursor(
+ rows=[{"id": 1, "url": "https://a.com/"}, {"id": 2, "url": "https://a.com"}]
+ ),
+ "SELECT url, data FROM lighthouse_page_summaries": _Cursor(
+ rows=[{"url": "https://a.com", "data": {"p": 1}}, {"url": "https://b.com", "data": "x"}]
+ ),
+ }
+ )
+ by_url = read_lh_runs_by_url(conn) # type: ignore[arg-type]
+ assert by_url["https://a.com"] == [1, 2]
+ summaries = read_lighthouse_page_summaries(conn) # type: ignore[arg-type]
+ assert summaries == {"https://a.com": {"p": 1}}
+
+
+def test_read_lh_audits_with_items_builds_details():
+ from website_profiling.db.lighthouse_store import read_lh_audits_with_items
+
+ conn = _Conn(
+ {
+ "SELECT * FROM lh_audits WHERE run_id": _Cursor(
+ rows=[
+ {
+ "id": 7,
+ "audit_id": "a11y",
+ "category_id": "accessibility",
+ "title": "T",
+ "description": "D",
+ "score": 0.8,
+ "score_display_mode": "binary",
+ "display_value": "ok",
+ "numeric_value": 1,
+ "help_text": "",
+ "details_type": "table",
+ "details_headings": [{"k": "v"}],
+ "details_meta": {"m": 1},
+ }
+ ]
+ ),
+ "SELECT row_data FROM lh_audit_items WHERE audit_row_id": _Cursor(rows=[{"row_data": {"x": 1}}]),
+ }
+ )
+ out = read_lh_audits_with_items(conn, 1) # type: ignore[arg-type]
+ assert out[0]["id"] == "a11y"
+ assert out[0]["details"]["items"][0]["x"] == 1
+
+
+def test_crawl_store_core_helpers(monkeypatch):
+ from website_profiling.db import crawl_store as cs
+
+ conn = _Conn(
+ {
+ "RETURNING id": _Cursor(row={"id": 11}),
+ "ORDER BY id DESC LIMIT 1": _Cursor(row={"id": 9}),
+ "WHERE id = %s": _Cursor(row={"created_at": "now", "start_url": "https://a.com"}),
+ }
+ )
+ assert cs.create_crawl_run(conn, "https://a.com") == 11 # type: ignore[arg-type]
+ assert cs.get_latest_crawl_run_id(conn) == 9 # type: ignore[arg-type]
+ info = cs.get_crawl_run_info(conn, 9) # type: ignore[arg-type]
+ assert info and info["start_url"] == "https://a.com"
+ assert cs._extract_hostname("https://A.com/path") == "a.com"
+
+
+def test_write_nodes_variants(monkeypatch):
+ from website_profiling.db import crawl_store as cs
+
+ conn = _Conn()
+ # empty df + run_id none => delete all nodes
+ cs.write_nodes(conn, pd.DataFrame(), crawl_run_id=None) # type: ignore[arg-type]
+ assert conn.commits >= 1
+
+ # df without required cols returns early
+ before = conn.commits
+ cs.write_nodes(conn, pd.DataFrame([{"x": 1}]), crawl_run_id=1) # type: ignore[arg-type]
+ assert conn.commits == before
+
+ # normal insert path with index->url rename
+ captured = {"n": 0}
+ monkeypatch.setattr(cs, "_executemany", lambda *_a, **_k: captured.__setitem__("n", captured["n"] + 1))
+ df = pd.DataFrame([{"index": "https://a.com", "count": 3}])
+ cs.write_nodes(conn, df, crawl_run_id=2) # type: ignore[arg-type]
+ assert captured["n"] == 1
+
+
+def test_historical_read_and_restore(monkeypatch):
+ from website_profiling.db import historical as h
+
+ class HistCursor:
+ def __init__(self):
+ self.rows = [{"id": 1}]
+
+ def execute(self, _sql):
+ return None
+
+ def fetchall(self):
+ return self.rows
+
+ class HistConn:
+ def __init__(self):
+ self.commits = 0
+
+ def cursor(self):
+ cur = HistCursor()
+
+ class CM:
+ def __enter__(self_non):
+ return cur
+
+ def __exit__(self_non, _t, _v, _tb):
+ return False
+
+ return CM()
+
+ def execute(self, _sql, _p=None):
+ return None
+
+ def commit(self):
+ self.commits += 1
+
+ class Ctx:
+ def __enter__(self):
+ return HistConn()
+
+ def __exit__(self, _t, _v, _tb):
+ return False
+
+ monkeypatch.setattr(h, "db_session", lambda: Ctx())
+ data = h.read_historical_data()
+ assert "report_payload" in data
+
+ conn = HistConn()
+ monkeypatch.setattr(h, "_executemany", lambda *_a, **_k: None)
+ h.restore_historical_data(
+ conn, # type: ignore[arg-type]
+ {
+ "report_payload": [{"id": 1, "generated_at": "x", "site_name": "s", "canonical_domain": "d", "data": {}}],
+ "lighthouse_summary": [{"id": 1, "created_at": "x", "data": {}}],
+ "lighthouse_runs": [{"id": 1, "created_at": "x", "url": "u", "strategy": "mobile", "run_index": 0, "data": {}}],
+ "lighthouse_page_summaries": [{"url": "u", "created_at": "x", "data": {}}],
+ "lh_audits": [],
+ "lh_audit_items": [],
+ "google_data": [],
+ "keyword_data": [],
+ "keyword_history": [],
+ "keyword_suggest_cache": [],
+ "crawl_runs": [{"id": 1, "created_at": "x", "start_url": "u"}],
+ },
+ )
+ assert conn.commits == 1
+
+
+def test_lighthouse_store_write_audits_from_run(monkeypatch):
+ from website_profiling.db import lighthouse_store as ls
+
+ conn = _Conn({"SELECT id FROM lh_audits": _Cursor(rows=[{"id": 100}])})
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.lighthouse.schema",
+ types.SimpleNamespace(
+ lhr_to_audit_rows=lambda _d: (
+ [
+ {
+ "audit_id": "a",
+ "category_id": "c",
+ "score": 1,
+ "score_display_mode": "binary",
+ "title": "t",
+ "description": "d",
+ "display_value": "",
+ "numeric_value": 1,
+ "help_text": "",
+ "details_type": "table",
+ "details_headings": "[]",
+ "details_meta": "{}",
+ }
+ ],
+ [(0, 0, {"x": 1})],
+ )
+ ),
+ )
+ calls = {"n": 0}
+ monkeypatch.setattr(ls, "_executemany", lambda *_a, **_k: calls.__setitem__("n", calls["n"] + 1))
+ ls.write_lh_audits_from_run(conn, 1, {}) # type: ignore[arg-type]
+ assert calls["n"] >= 1
+
+
+def test_google_run_google_test_paths(monkeypatch):
+ from website_profiling.commands import google_cmd
+
+ import sys as _sys
+ import types as _types
+
+ google_mod = _types.ModuleType("google")
+ auth_mod = _types.ModuleType("google.auth")
+ exc_mod = _types.ModuleType("google.auth.exceptions")
+ exc_mod.RefreshError = RuntimeError
+ auth_mod.exceptions = exc_mod
+ google_mod.auth = auth_mod
+ monkeypatch.setitem(_sys.modules, "google", google_mod)
+ monkeypatch.setitem(_sys.modules, "google.auth", auth_mod)
+ monkeypatch.setitem(_sys.modules, "google.auth.exceptions", exc_mod)
+
+ monkeypatch.setitem(
+ _sys.modules,
+ "website_profiling.integrations.google.auth",
+ _types.SimpleNamespace(
+ build_credentials=lambda *_a, **_k: object(),
+ read_secrets=lambda *_a, **_k: {"gscSiteUrl": "", "ga4PropertyId": ""},
+ ),
+ )
+ # no ids configured => warnings => exit 1
+ with pytest.raises(SystemExit) as e:
+ google_cmd._run_google_test(None)
+ assert e.value.code == 1
+
+
diff --git a/tests/test_pipeline_cmd_run_unit.py b/tests/test_pipeline_cmd_run_unit.py
new file mode 100644
index 00000000..314fe2c6
--- /dev/null
+++ b/tests/test_pipeline_cmd_run_unit.py
@@ -0,0 +1,71 @@
+import argparse
+import types
+
+import pandas as pd
+
+
+def test_pipeline_run_calls_crawl_with_minimal_config(monkeypatch) -> None:
+ from website_profiling.commands import pipeline_cmd
+
+ called = {"crawl": 0}
+
+ def fake_run_crawler(**_kwargs):
+ called["crawl"] += 1
+ return pd.DataFrame([{"url": "https://site.com", "status": 200}])
+
+ # Patch crawler module function used in _run_crawl
+ import website_profiling.crawl.crawler as crawler_mod
+
+ monkeypatch.setattr(crawler_mod, "run_crawler", fake_run_crawler)
+
+ cfg = {
+ "start_url": "https://site.com",
+ "run_crawl": "true",
+ "run_report": "false",
+ "run_plot": "false",
+ "run_lighthouse": "false",
+ "run_lighthouse_on_pages": "false",
+ }
+ args = argparse.Namespace(command=None)
+ pipeline_cmd.run(cfg, args)
+ assert called["crawl"] == 1
+
+
+def test_pipeline_lighthouse_on_pages_uses_selected_urls(monkeypatch) -> None:
+ from website_profiling.commands import pipeline_cmd
+
+ # Fake db session / read_crawl
+ class _Ctx:
+ def __enter__(self):
+ return object()
+
+ def __exit__(self, _t, _v, _tb):
+ return False
+
+ import website_profiling.db as db
+
+ monkeypatch.setattr(db, "db_session", lambda: _Ctx())
+ monkeypatch.setattr(db, "get_latest_crawl_run_id", lambda _c: 1)
+ monkeypatch.setattr(
+ db,
+ "read_crawl",
+ lambda _c, _rid: pd.DataFrame(
+ [{"url": "https://a.com", "status": 200}, {"url": "https://b.com", "status": 404}]
+ ),
+ )
+
+ urls_seen = {}
+
+ def fake_lh_on_pages(urls, **_kwargs):
+ urls_seen["urls"] = urls
+
+ monkeypatch.setitem(
+ __import__("sys").modules,
+ "website_profiling.lighthouse.runner",
+ types.SimpleNamespace(run_lighthouse_on_pages=fake_lh_on_pages),
+ )
+
+ cfg = {"lighthouse_strategy": "mobile", "lighthouse_iterations": "1"}
+ pipeline_cmd._run_lighthouse_on_pages(cfg, lighthouse_max_pages=10)
+ assert urls_seen["urls"] == ["https://a.com"]
+
diff --git a/tests/test_pipeline_lighthouse_url_selection.py b/tests/test_pipeline_lighthouse_url_selection.py
new file mode 100644
index 00000000..16a9e91f
--- /dev/null
+++ b/tests/test_pipeline_lighthouse_url_selection.py
@@ -0,0 +1,35 @@
+import pandas as pd
+
+
+def test_select_lighthouse_urls_from_crawl_empty_df_is_safe() -> None:
+ from website_profiling.commands.pipeline_cmd import select_lighthouse_urls_from_crawl
+
+ assert select_lighthouse_urls_from_crawl(pd.DataFrame(), max_pages=10) == []
+
+
+def test_select_lighthouse_urls_from_crawl_requires_url_column() -> None:
+ from website_profiling.commands.pipeline_cmd import select_lighthouse_urls_from_crawl
+
+ df = pd.DataFrame([{"status": "200"}])
+ assert select_lighthouse_urls_from_crawl(df, max_pages=10) == []
+
+
+def test_select_lighthouse_urls_from_crawl_filters_to_2xx_and_dedupes() -> None:
+ from website_profiling.commands.pipeline_cmd import select_lighthouse_urls_from_crawl
+
+ df = pd.DataFrame(
+ [
+ {"url": "https://a.com", "status": 200},
+ {"url": "https://a.com", "status": "200"},
+ {"url": "https://b.com", "status": 204},
+ {"url": "https://c.com", "status": 404},
+ {"url": None, "status": 200},
+ {"url": " https://d.com ", "status": 201},
+ ]
+ )
+ assert select_lighthouse_urls_from_crawl(df, max_pages=3) == [
+ "https://a.com",
+ "https://b.com",
+ "https://d.com",
+ ]
+
diff --git a/tests/test_pool_and_report_store_unit.py b/tests/test_pool_and_report_store_unit.py
new file mode 100644
index 00000000..f83f2494
--- /dev/null
+++ b/tests/test_pool_and_report_store_unit.py
@@ -0,0 +1,34 @@
+import os
+import types
+
+from tests.db_test_fakes import FakeConn
+
+
+def test_get_database_url_appends_connect_timeout(monkeypatch) -> None:
+ from website_profiling.db import pool
+
+ monkeypatch.setenv("DATABASE_URL", "postgres://u:p@localhost:5432/db")
+ assert "connect_timeout=" in pool.get_database_url()
+
+ monkeypatch.setenv("DATABASE_URL", "postgres://u:p@localhost:5432/db?connect_timeout=1")
+ assert pool.get_database_url().count("connect_timeout=") == 1
+
+
+def test_canonical_domain_from_report_prefers_start_url(monkeypatch) -> None:
+ from website_profiling.db import report_store
+
+ # Patch get_crawl_run_info used inside report_store
+ monkeypatch.setattr(report_store, "get_crawl_run_info", lambda _c, _rid: {"start_url": "https://Start.Example/path"})
+ conn = FakeConn()
+ domain = report_store._canonical_domain_from_report(conn, {"crawl_run_id": 1, "top_pages": [{"url": "https://x.com"}]}) # type: ignore[arg-type]
+ assert domain == "start.example"
+
+
+def test_canonical_domain_falls_back_to_top_pages(monkeypatch) -> None:
+ from website_profiling.db import report_store
+
+ monkeypatch.setattr(report_store, "get_crawl_run_info", lambda _c, _rid: None)
+ conn = FakeConn()
+ domain = report_store._canonical_domain_from_report(conn, {"top_pages": [{"url": "https://Top.Example/a"}]}) # type: ignore[arg-type]
+ assert domain == "top.example"
+
diff --git a/tests/test_small_coverage_win.py b/tests/test_small_coverage_win.py
new file mode 100644
index 00000000..094b77b2
--- /dev/null
+++ b/tests/test_small_coverage_win.py
@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import types
+
+
+class Conn:
+ def __init__(self, row=None, rows=None, boom=False):
+ self.row = row
+ self.rows = rows or []
+ self.boom = boom
+ self.commits = 0
+ self.executed = []
+
+ def execute(self, sql, params=None):
+ self.executed.append((sql, params))
+ if self.boom:
+ raise RuntimeError("boom")
+ return self
+
+ def fetchone(self):
+ return self.row
+
+ def fetchall(self):
+ return list(self.rows)
+
+ def commit(self):
+ self.commits += 1
+
+ def transaction(self):
+ class CM:
+ def __enter__(self_non):
+ return None
+
+ def __exit__(self_non, _t, _v, _tb):
+ return False
+
+ return CM()
+
+
+def test_config_store_error_fallbacks():
+ from website_profiling.db.config_store import read_llm_config, read_pipeline_config
+
+ known, unknown = read_pipeline_config(Conn(boom=True)) # type: ignore[arg-type]
+ assert known == {}
+ assert unknown == []
+ assert read_llm_config(Conn(boom=True)) == {} # type: ignore[arg-type]
+
+
+def test_config_store_write_llm_config_commits():
+ from website_profiling.db.config_store import write_llm_config
+
+ conn = Conn()
+ write_llm_config(conn, {"k": "v"}, secret_keys={"k"}) # type: ignore[arg-type]
+ assert conn.commits == 0 # transaction style; no explicit commit in function
+ assert any("INSERT INTO llm_config" in s for s, _ in conn.executed)
+
+
+def test_llm_cache_write_and_error_paths():
+ from website_profiling.db.llm_cache_store import read_llm_cache_batch, write_llm_cache
+
+ conn = Conn()
+ write_llm_cache(conn, "k", '{"a":1}') # type: ignore[arg-type]
+ write_llm_cache(conn, "k", "not-json") # type: ignore[arg-type]
+ assert conn.commits == 2
+
+ # exception branch
+ assert read_llm_cache_batch(Conn(boom=True), ["k"]) == {} # type: ignore[arg-type]
+
+
+def test_report_store_write_and_read_none():
+ from website_profiling.db.report_store import _extract_hostname, read_report_payload, write_report_payload
+
+ conn = Conn(row=None)
+ assert read_report_payload(conn) is None # type: ignore[arg-type]
+ conn2 = Conn()
+ write_report_payload(conn2, {"site_name": "S", "top_pages": [{"url": "https://x.com"}]}) # type: ignore[arg-type]
+ assert conn2.commits == 1
+ assert _extract_hostname("https://X.com/a") == "x.com"
+
diff --git a/tests/test_storage_bulk.py b/tests/test_storage_bulk.py
new file mode 100644
index 00000000..f4c928cb
--- /dev/null
+++ b/tests/test_storage_bulk.py
@@ -0,0 +1,53 @@
+"""Bulk PostgreSQL storage tests."""
+from __future__ import annotations
+
+import os
+
+import pandas as pd
+import pytest
+from psycopg.types.json import Json
+
+from website_profiling.db import db_session, read_crawl, write_crawl
+from website_profiling.db.storage import create_crawl_run, write_crawl_batch
+
+
+@pytest.fixture(scope="module")
+def pg_conn():
+ if not (os.environ.get("DATABASE_URL") or "").strip():
+ pytest.skip("DATABASE_URL not set — start Postgres and run alembic upgrade head")
+ with db_session() as conn:
+ yield conn
+
+
+def test_write_crawl_bulk_round_trip(pg_conn):
+ run_id = create_crawl_run(pg_conn, "https://example.com")
+ rows = []
+ for i in range(100):
+ rows.append(
+ (
+ run_id,
+ f"https://example.com/page-{i}",
+ "200",
+ f"Page {i}",
+ Json({"status": "200", "title": f"Page {i}"}),
+ )
+ )
+ write_crawl_batch(pg_conn, rows, run_id, commit=True)
+
+ df = read_crawl(pg_conn, run_id)
+ assert len(df) == 100
+ assert "url" in df.columns
+ assert df["status"].astype(str).str.startswith("200").all()
+
+
+def test_write_crawl_dataframe_executemany(pg_conn):
+ run_id = create_crawl_run(pg_conn, "https://bulk.example.com")
+ data = {
+ "url": [f"https://bulk.example.com/{i}" for i in range(50)],
+ "status": ["200"] * 50,
+ "title": [f"T{i}" for i in range(50)],
+ }
+ df = pd.DataFrame(data)
+ write_crawl(pg_conn, df, crawl_run_id=run_id)
+ out = read_crawl(pg_conn, run_id)
+ assert len(out) == 50
diff --git a/tests/test_suggest_cache_concurrent.py b/tests/test_suggest_cache_concurrent.py
new file mode 100644
index 00000000..5b654c12
--- /dev/null
+++ b/tests/test_suggest_cache_concurrent.py
@@ -0,0 +1,55 @@
+"""Thread-safe suggest cache tests."""
+from __future__ import annotations
+
+import os
+from unittest.mock import patch
+
+import pytest
+
+from website_profiling.db import db_session
+from website_profiling.integrations.google.suggest import batch_expand, flush_suggest_cache
+
+
+@pytest.fixture(scope="module")
+def pg_conn():
+ if not (os.environ.get("DATABASE_URL") or "").strip():
+ pytest.skip("DATABASE_URL not set — start Postgres and run alembic upgrade head")
+ with db_session() as conn:
+ yield conn
+
+
+def _fake_fetch(task, timeout=8.0):
+ seed, source, lang, country = task
+ return seed, source, [f"{seed}-{source}-a", f"{seed}-{source}-b"]
+
+
+def test_batch_expand_flushes_cache_on_main_thread(pg_conn):
+ seeds = [f"seed{i}" for i in range(6)]
+ with patch(
+ "website_profiling.integrations.google.suggest._fetch_one",
+ side_effect=_fake_fetch,
+ ):
+ result = batch_expand(
+ seeds,
+ sources=("web",),
+ max_workers=4,
+ cache_conn=pg_conn,
+ )
+ assert len(result) == 6
+ for seed in seeds:
+ assert result[seed]["web"]
+
+ cur = pg_conn.execute(
+ "SELECT COUNT(*) AS n FROM keyword_suggest_cache WHERE cache_key LIKE 'web:%'"
+ )
+ row = cur.fetchone()
+ assert int(row["n"]) >= 6
+
+
+def test_flush_suggest_cache_executemany(pg_conn):
+ entries = [(f"bulk{i}", "web", [f"kw{i}"]) for i in range(10)]
+ flush_suggest_cache(pg_conn, entries)
+ cur = pg_conn.execute(
+ "SELECT COUNT(*) AS n FROM keyword_suggest_cache WHERE cache_key LIKE 'web:bulk%'"
+ )
+ assert int(cur.fetchone()["n"]) >= 10
diff --git a/web/app/(reports)/[slug]/page.tsx b/web/app/(reports)/[slug]/page.tsx
new file mode 100644
index 00000000..62bc1493
--- /dev/null
+++ b/web/app/(reports)/[slug]/page.tsx
@@ -0,0 +1,18 @@
+import ReportShell from '@/ReportShell';
+import { pathSlugToViewId } from '@/routes';
+import { notFound } from 'next/navigation';
+import type { ReactElement } from 'react';
+
+export const dynamic = 'force-dynamic';
+
+export default async function SlugPage({
+ params,
+}: {
+ params: Promise<{ slug: string }>;
+}): Promise+ That report view does not exist. Choose a valid section from the app. +
+{strings.app.loading}
-- {isDomainError ? strings.app.noReportForDomainTitle : strings.app.failedTitle} -
-{error}
- {!isDomainError ? ( -{strings.app.failedHint}
- ) : null} - -+ {isDomainError ? strings.app.noReportForDomainTitle : strings.app.failedTitle} +
+{error}
+ {!isDomainError ? ( +{strings.app.failedHint}
+ ) : null} + + Open Pipeline + +{description}
: null} +{helper}
} ++ Connect Search Console and GA4, then choose properties to sync with your reports. +
+
+ Need a project?{' '}
+
+ Google Cloud guide
Account connected
+You can configure properties in the next step.
+Complete step 1 to enable sign-in.
+ ) : ( + + )} + +Load sites from your connected account.
+ +{properties.ga4ListError}
+ ) : ( +Numeric ID from GA4 Admin → Property settings.
+ )} ++ Last fetched: {new Date(status.lastFetchedAt).toLocaleString()} +
+ ) : null} + +
+ {testLog}
+
+ ) : null}
+
+ {fetchLog ? (
+
+ Fetch status:{' '}
+
+ {fetchJobStatus}
+
+ {fetchJobStatus === 'running' ? (
+
+ {fetchLog}
+
+ {helper}
} -
- You need a{' '}
-
- Google Cloud project
- Complete Step 1 first to enable this button. -
- )} -- {properties.ga4ListError} -
- )} -- Numeric ID from GA4 Admin > Property Settings (not the G-XXXXXXX Measurement ID). -
-
- {testLog}
-
- )}
- {fetchLog && (
-
- Status:{' '}
-
- {fetchJobStatus}
-
- {fetchJobStatus === 'running' && (
-
-
- {fetchLog}
-
- - Full live log also opens in Pipeline Runner (blue terminal button, bottom-right). -
-- Last fetched: {new Date(status.lastFetchedAt).toLocaleString()} -
- )} - - {/* ── Advanced: paste refresh token ── */} -{subtitle}
} -{f.help}
- ) : null; - - if (f.type === 'bool') { - const checked = value === true; - return ( -