diff --git a/pyproject.toml b/pyproject.toml index aa3b455..a226bc6 100644 --- a/pyproject.toml +++ b/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 diff --git a/tests/dag/__init__.py b/tests/dag/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/dag/test_dags.py b/tests/dag/test_dags.py new file mode 100644 index 0000000..cd9df02 --- /dev/null +++ b/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})" + ) diff --git a/tests/integration/test_pipeline.py b/tests/integration/test_pipeline.py deleted file mode 100644 index 91ff378..0000000 --- a/tests/integration/test_pipeline.py +++ /dev/null @@ -1,130 +0,0 @@ -"""Integration test: dlt normalizes the fixtures into the DuckDB `raw` schema. - -Feeds the recorded API fixtures through dlt (using the real `stamp` metadata helper) -into a temporary DuckDB and asserts the normalized table/column contract that the dbt -staging models depend on. Also checks idempotency of the merge write disposition. -""" - -import json -from collections.abc import Iterator -from pathlib import Path - -import dlt -import duckdb -import pytest - -from repolytics.ingestion._meta import stamp - -FIXTURES = Path(__file__).parent.parent / "fixtures" -REPO = "encode/httpx" -PACKAGE = "httpx" # matches the repo fixture so the downloads->repo FK resolves - - -def _read(rel: str) -> object: - return json.loads((FIXTURES / rel).read_text(encoding="utf-8")) - - -def _sources() -> list: - @dlt.resource(name="repositories", write_disposition="merge", primary_key="id") - def repositories() -> Iterator[dict]: - yield stamp(_repo=REPO)(_read("github_responses/repository.json")) - - @dlt.resource(name="commits", write_disposition="merge", primary_key="sha") - def commits() -> Iterator[dict]: - rows = _read("github_responses/commits.json") - yield from (stamp(_repo=REPO)(c) for c in rows) - - @dlt.resource( - name="issues", write_disposition="merge", primary_key=["_repo", "number"] - ) - def issues() -> Iterator[dict]: - yield from (stamp(_repo=REPO)(c) for c in _read("github_responses/issues.json")) - - @dlt.resource( - name="pull_requests", write_disposition="merge", primary_key=["_repo", "number"] - ) - def pull_requests() -> Iterator[dict]: - yield from (stamp(_repo=REPO)(c) for c in _read("github_responses/pulls.json")) - - @dlt.resource(name="releases", write_disposition="merge", primary_key="id") - def releases() -> Iterator[dict]: - yield from ( - stamp(_repo=REPO)(c) for c in _read("github_responses/releases.json") - ) - - @dlt.resource( - name="downloads", - write_disposition="merge", - primary_key=["_package", "category", "date"], - ) - def downloads() -> Iterator[dict]: - data = _read("pypi_responses/overall.json")["data"] - yield from (stamp(_package=PACKAGE)(r) for r in data) - - return [repositories, commits, issues, pull_requests, releases, downloads] - - -@pytest.fixture -def loaded_db(tmp_duckdb_path: Path) -> Iterator[duckdb.DuckDBPyConnection]: - pipeline = dlt.pipeline( - pipeline_name="repolytics_test", - destination=dlt.destinations.duckdb(str(tmp_duckdb_path)), - dataset_name="raw", - ) - pipeline.run(_sources()) - 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_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 diff --git a/tests/integration/test_raw_contract.py b/tests/integration/test_raw_contract.py new file mode 100644 index 0000000..bd52376 --- /dev/null +++ b/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 diff --git a/tests/support/__init__.py b/tests/support/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/support/warehouse.py b/tests/support/warehouse.py new file mode 100644 index 0000000..6b029dd --- /dev/null +++ b/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}") diff --git a/tests/unit/test_pipeline.py b/tests/unit/test_pipeline.py new file mode 100644 index 0000000..b2e31c2 --- /dev/null +++ b/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) == {} diff --git a/uv.lock b/uv.lock index 6f12716..fdb2b70 100644 --- a/uv.lock +++ b/uv.lock @@ -2378,6 +2378,7 @@ dbt = [ dev = [ { name = "pytest" }, { name = "pytest-cov" }, + { name = "responses" }, { name = "ruff" }, ] orchestration = [ @@ -2399,6 +2400,7 @@ dbt = [ dev = [ { name = "pytest", specifier = ">=9.1.0" }, { name = "pytest-cov", specifier = ">=7.1.0" }, + { name = "responses", specifier = ">=0.26.1" }, { name = "ruff", specifier = ">=0.15.17" }, ] orchestration = [{ name = "astronomer-cosmos", specifier = ">=1.14.2" }] @@ -2430,6 +2432,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/bd/60/50fbb6ffb35f733654466f1a90d162bcbea358adc3b0871339254fbc37b2/requirements_parser-0.13.0-py3-none-any.whl", hash = "sha256:2b3173faecf19ec5501971b7222d38f04cb45bb9d87d0ad629ca71e2e62ded14", size = 14782, upload-time = "2025-05-21T13:42:04.007Z" }, ] +[[package]] +name = "responses" +version = "0.26.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pyyaml" }, + { name = "requests" }, + { name = "urllib3" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c2/58/1fb6de3503428196df78638f991ec8095274f1ee9723e272ee4d9ff0092b/responses-0.26.1.tar.gz", hash = "sha256:2eb3218553cc8f79b57d257bac23af5e1bf381f5b9390b1767816f0843e01dc2", size = 83088, upload-time = "2026-05-21T19:56:39.747Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/3a/31/6a620b4427d546b9e7cca8b3b8c5f0559d9cef2bb9eedcda7f73c1473c19/responses-0.26.1-py3-none-any.whl", hash = "sha256:8aacc4586eb08fb2208ef64a9eb4258d9b0c6e6f4260845f2f018ab847495345", size = 35502, upload-time = "2026-05-21T19:56:38.046Z" }, +] + [[package]] name = "rich" version = "15.0.0"