test: cover ingestion sources and DAG, add branch coverage

pyproject.toml

@@ -21,6 +21,7 @@ orchestration = [
 dev = [
     "pytest>=9.1.0",
     "pytest-cov>=7.1.0",
+    "responses>=0.26.1",
     "ruff>=0.15.17",
 ]
 
@@ -42,3 +43,4 @@ xfail_strict = true
 
 [tool.coverage.run]
 source = ["repolytics"]
+branch = true

tests/dag/test_dags.py

@@ -0,0 +1,38 @@
+"""DAG tests: generic policy checks over every DAG."""
+
+import os
+from pathlib import Path
+
+import pytest
+from airflow.dag_processing.dagbag import DagBag
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+DAGS_DIR = REPO_ROOT / "dags"
+MANIFEST = REPO_ROOT / "dbt" / "target" / "manifest.json"
+
+if not MANIFEST.exists():
+    raise RuntimeError(f"dbt manifest not found at {MANIFEST} - run `dbt parse` first.")
+
+# Cosmos reads this at DAG import time; set it before building the bag.
+os.environ.setdefault("DBT_PROJECT_DIR", str(REPO_ROOT / "dbt"))
+
+DAG_BAG = DagBag(dag_folder=str(DAGS_DIR), include_examples=False)
+ALL_DAGS = list(DAG_BAG.dags.values())
+DAG_IDS = [dag.dag_id for dag in ALL_DAGS]
+
+
+def test_no_import_errors() -> None:
+    assert DAG_BAG.import_errors == {}, DAG_BAG.import_errors
+
+
+@pytest.mark.parametrize("dag", ALL_DAGS, ids=DAG_IDS)
+def test_dag_is_tagged(dag) -> None:
+    assert dag.tags, f"{dag.dag_id} has no tags"
+
+
+@pytest.mark.parametrize("dag", ALL_DAGS, ids=DAG_IDS)
+def test_dag_retries_at_least_two(dag) -> None:
+    retries = dag.default_args.get("retries")
+    assert retries is not None and retries >= 2, (
+        f"{dag.dag_id} must set retries >= 2 (got {retries!r})"
+    )

tests/integration/test_raw_contract.py

@@ -0,0 +1,90 @@
+"""Integration test: the real sources normalize fixtures into the DuckDB `raw` schema.
+
+Drives the production `github_source` / `pypi_source` (HTTP mocked from the recorded
+fixtures via `tests.support.warehouse.load_raw_fixtures`) into a temporary DuckDB and
+asserts the normalized table/column contract that the dbt staging models depend on.
+"""
+
+from collections.abc import Iterator
+from pathlib import Path
+
+import duckdb
+import pytest
+
+from tests.support.warehouse import PACKAGE, REPO, load_raw_fixtures
+
+
+@pytest.fixture
+def loaded_db(tmp_duckdb_path: Path) -> Iterator[duckdb.DuckDBPyConnection]:
+    load_raw_fixtures(tmp_duckdb_path)
+    conn = duckdb.connect(str(tmp_duckdb_path))
+    try:
+        yield conn
+    finally:
+        conn.close()
+
+
+def _tables(conn: duckdb.DuckDBPyConnection) -> set[str]:
+    rows = conn.execute(
+        "select table_name from information_schema.tables where table_schema = 'raw'"
+    ).fetchall()
+    return {r[0] for r in rows}
+
+
+def test_normalized_tables_and_child_tables_exist(
+    loaded_db: duckdb.DuckDBPyConnection,
+) -> None:
+    assert {
+        "repositories",
+        "repositories__topics",
+        "commits",
+        "commits__parents",
+        "issues",
+        "issues__labels",
+        "pull_requests",
+        "pull_requests__labels",
+        "releases",
+        "downloads",
+    } <= _tables(loaded_db)
+
+
+def test_metadata_columns_present(loaded_db: duckdb.DuckDBPyConnection) -> None:
+    commit_cols = {
+        r[0]
+        for r in loaded_db.execute(
+            "select column_name from information_schema.columns "
+            "where table_schema = 'raw' and table_name = 'commits'"
+        ).fetchall()
+    }
+    assert {"_repo", "_loaded_at", "commit__author__date"} <= commit_cols
+
+    repo = loaded_db.execute("select distinct _repo from raw.commits").fetchone()[0]
+    assert repo == REPO
+    package = loaded_db.execute(
+        "select distinct _package from raw.downloads"
+    ).fetchone()[0]
+    assert package == PACKAGE
+
+
+def test_pull_request_marker_column_exists(
+    loaded_db: duckdb.DuckDBPyConnection,
+) -> None:
+    # The github source forces `pull_request__url` to exist so the staging PR filter
+    # never references a missing column - assert the real source still does so.
+    issue_cols = {
+        r[0]
+        for r in loaded_db.execute(
+            "select column_name from information_schema.columns "
+            "where table_schema = 'raw' and table_name = 'issues'"
+        ).fetchall()
+    }
+    assert "pull_request__url" in issue_cols
+
+
+def test_row_counts(loaded_db: duckdb.DuckDBPyConnection) -> None:
+    def count(table: str) -> int:
+        return loaded_db.execute(f"select count(*) from raw.{table}").fetchone()[0]
+
+    assert count("commits") == 2
+    assert count("issues") == 2  # includes the PR-shaped issue (staging filters it)
+    assert count("downloads") == 2

tests/support/warehouse.py

@@ -0,0 +1,67 @@
+"""Load the recorded API fixtures into a DuckDB `raw` dataset via the real sources.
+
+Run standalone (used by CI to seed a warehouse before `dbt build`)::
+    DUCKDB_PATH=/tmp/ci.duckdb python -m tests.support.warehouse
+"""
+
+import json
+import os
+from pathlib import Path
+
+import dlt
+import responses
+
+from repolytics.ingestion.github_source import github_source
+from repolytics.ingestion.pypi_source import pypi_source
+
+FIXTURES = Path(__file__).parent.parent / "fixtures"
+REPO = "encode/httpx"
+PACKAGE = "httpx"
+
+GITHUB_API = "https://api.github.com"
+PYPI_API = "https://pypistats.org/api"
+
+
+def _json(rel: str) -> object:
+    return json.loads((FIXTURES / rel).read_text(encoding="utf-8"))
+
+
+def _register(rsps: responses.RequestsMock) -> None:
+    """Stub every endpoint the real sources hit, one single-page response each.
+
+    No `Link` header is set, so dlt's `RESTClient.paginate` treats each list
+    response as a single page and stops.
+    """
+    owner, name = REPO.split("/")
+    base = f"{GITHUB_API}/repos/{owner}/{name}"
+    rsps.get(base, json=_json("github_responses/repository.json"))
+    rsps.get(f"{base}/commits", json=_json("github_responses/commits.json"))
+    rsps.get(f"{base}/issues", json=_json("github_responses/issues.json"))
+    rsps.get(f"{base}/pulls", json=_json("github_responses/pulls.json"))
+    rsps.get(f"{base}/releases", json=_json("github_responses/releases.json"))
+    rsps.get(
+        f"{PYPI_API}/packages/{PACKAGE}/overall",
+        json=_json("pypi_responses/overall.json"),
+    )
+
+
+def load_raw_fixtures(duckdb_path: Path) -> None:
+    """Run the real GitHub + PyPI sources against the fixtures into `raw`."""
+    pipeline = dlt.pipeline(
+        pipeline_name="repolytics_fixtures",
+        destination=dlt.destinations.duckdb(str(duckdb_path)),
+        dataset_name="raw",
+    )
+    # assert_all_requests_are_fired=False: dlt may not re-hit every stub on a no-op
+    # pass, and we only care that the calls it does make are answered from fixtures.
+    with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
+        _register(rsps)
+        pipeline.run(github_source([REPO], token="fixture-token"))
+        pipeline.run(pypi_source([PACKAGE], min_interval=0))
+
+
+if __name__ == "__main__":
+    path = Path(os.environ["DUCKDB_PATH"])
+    path.parent.mkdir(parents=True, exist_ok=True)
+    load_raw_fixtures(path)
+    print(f"Loaded fixtures into raw dataset at {path}")

tests/unit/test_pipeline.py

@@ -0,0 +1,53 @@
+"""Unit tests for repolytics.ingestion.pipeline helpers and skip paths."""
+
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from repolytics.config import Settings
+from repolytics.ingestion import pipeline
+
+
+def test_table_row_counts_filters_dlt_internal_tables() -> None:
+    fake = SimpleNamespace(
+        last_trace=SimpleNamespace(
+            last_normalize_info=SimpleNamespace(
+                row_counts={
+                    "commits": 2,
+                    "issues": 3,
+                    "_dlt_loads": 1,
+                    "_dlt_pipeline_state": 1,
+                }
+            )
+        )
+    )
+
+    assert pipeline._table_row_counts(fake) == {"commits": 2, "issues": 3}
+
+
+def test_run_github_skips_when_no_repos(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setenv("GITHUB_TOKEN", "ghp_secret")
+    # build_pipeline would hit the network/DuckDB; assert it is never reached.
+    monkeypatch.setattr(
+        pipeline, "build_pipeline", lambda *_a, **_k: pytest.fail("should not run")
+    )
+    settings = Settings(_env_file=None, projects_file=tmp_path / "missing.csv")
+
+    assert settings.target_repos == []
+    assert pipeline.run_github(settings) == {}
+
+
+def test_run_pypi_skips_when_no_packages(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setenv("GITHUB_TOKEN", "ghp_secret")
+    monkeypatch.setattr(
+        pipeline, "build_pipeline", lambda *_a, **_k: pytest.fail("should not run")
+    )
+    settings = Settings(_env_file=None, projects_file=tmp_path / "missing.csv")
+
+    assert settings.packages == []
+    assert pipeline.run_pypi(settings) == {}