test_log_schema.py

"""Validates structured JSONL log files from the three ingestors."""
from __future__ import annotations

import json
from datetime import datetime
from pathlib import Path

import pytest

LOGS_DIR = Path(__file__).resolve().parents[1] / "data" / "logs"

LOG_FILES = {
    "github":         LOGS_DIR / "github_ingestor_structured.jsonl",
    "pypi":           LOGS_DIR / "pypi_ingestor_structured.jsonl",
    "stackoverflow":  LOGS_DIR / "stackoverflow_ingestor_structured.jsonl",
}

BASE_FIELDS = {"timestamp", "event", "package", "status"}

EVENT_SCHEMA: dict[str, set[str]] = {
    # GitHub
    "repo_metadata":    {"bytes", "stars", "forks", "open_issues"},
    "readme":           {"readme_chars", "readme_encoding"},
    "contributors":     {"contributor_count"},
    "run_complete":     {"success", "error"},
    # PyPI
    "package_metadata": {"bytes", "latest_version", "release_count"},
    "downloads_recent": {"last_day", "last_week", "last_month"},
    "downloads_overall":{"bytes", "total_downloads", "row_count"},
    # Stack Overflow
    "tag_info":         {"total_questions"},
    "questions":        {"bytes", "question_count", "answered_count"},
    "answers":          {"bytes", "answer_count"},
}


def _parse_jsonl(path: Path) -> list[dict]:
    with path.open("r", encoding="utf-8") as f:
        return [json.loads(line) for line in f if line.strip()]


@pytest.fixture(scope="module", params=list(LOG_FILES.keys()))
def log_data(request):
    source = request.param
    path = LOG_FILES[source]
    assert path.exists(), f"Log file missing: {path}"
    records = _parse_jsonl(path)
    assert len(records) > 0, f"Log file is empty: {path}"
    return source, records


class TestBaseSchema:
    def test_all_records_have_base_fields(self, log_data):
        _, records = log_data
        for rec in records:
            missing = BASE_FIELDS - rec.keys()
            assert not missing, f"Record missing base fields {missing}: {rec}"

    def test_timestamp_is_iso_parseable(self, log_data):
        _, records = log_data
        for rec in records:
            ts = rec["timestamp"]
            assert isinstance(ts, str), f"timestamp must be a string, got {type(ts)}"
            try:
                datetime.fromisoformat(ts.replace("Z", "+00:00"))
            except ValueError:
                pytest.fail(f"timestamp not ISO-parseable: {ts!r}")

    def test_status_values_are_valid(self, log_data):
        _, records = log_data
        valid = {"success", "error", "warning"}
        for rec in records:
            assert rec["status"] in valid, (
                f"Unexpected status {rec['status']!r} in: {rec}"
            )

    def test_package_field_is_string(self, log_data):
        _, records = log_data
        for rec in records:
            assert isinstance(rec["package"], str)


class TestEventSchema:
    def test_event_specific_fields_present(self, log_data):
        _, records = log_data
        for rec in records:
            if rec.get("status") != "success":
                continue  # error records won't have the data fields
            required = EVENT_SCHEMA.get(rec["event"], set())
            if not required:
                continue
            missing = required - rec.keys()
            assert not missing, (
                f"Event {rec['event']!r} missing fields {missing}: {rec}"
            )


class TestRunComplete:
    def test_each_source_has_at_least_one_run_complete(self, log_data):
        _, records = log_data
        run_completes = [r for r in records if r["event"] == "run_complete"]
        assert len(run_completes) >= 1, "No run_complete event found"

    def test_success_and_error_are_non_negative(self, log_data):
        _, records = log_data
        for rec in records:
            if rec["event"] == "run_complete":
                assert rec["success"] >= 0
                assert rec["error"] >= 0

    def test_eight_packages_tracked(self, log_data):
        _, records = log_data
        run_completes = [r for r in records if r["event"] == "run_complete"]
        for rc in run_completes:
            total = rc["success"] + rc["error"]
            assert total == 8, (
                f"Expected 8 packages per run, got {total} in: {rc}"
            )


class TestGitHubDataQuality:
    def test_stars_and_forks_non_negative(self, log_data):
        source, records = log_data
        if source != "github":
            pytest.skip("GitHub-only test")
        for rec in records:
            if rec["event"] == "repo_metadata" and rec["status"] == "success":
                assert rec["stars"] >= 0, f"Negative stars: {rec}"
                assert rec["forks"] >= 0, f"Negative forks: {rec}"
                assert rec["open_issues"] >= 0, f"Negative open_issues: {rec}"


class TestPyPIDataQuality:
    def test_download_counts_non_negative(self, log_data):
        source, records = log_data
        if source != "pypi":
            pytest.skip("PyPI-only test")
        for rec in records:
            if rec["event"] == "downloads_recent" and rec["status"] == "success":
                assert rec["last_day"] >= 0
                assert rec["last_week"] >= 0
                assert rec["last_month"] >= 0

    def test_weekly_le_monthly(self, log_data):
        source, records = log_data
        if source != "pypi":
            pytest.skip("PyPI-only test")
        for rec in records:
            if rec["event"] == "downloads_recent" and rec["status"] == "success":
                assert rec["last_week"] <= rec["last_month"], (
                    f"Weekly downloads exceed monthly for {rec['package']}: {rec}"
                )


class TestSODataQuality:
    def test_total_questions_positive(self, log_data):
        source, records = log_data
        if source != "stackoverflow":
            pytest.skip("SO-only test")
        for rec in records:
            if rec["event"] == "tag_info" and rec["status"] == "success":
                assert rec["total_questions"] > 0

    def test_answered_count_le_question_count(self, log_data):
        source, records = log_data
        if source != "stackoverflow":
            pytest.skip("SO-only test")
        for rec in records:
            if rec["event"] == "questions" and rec["status"] == "success":
                assert rec["answered_count"] <= rec["question_count"], (
                    f"answered_count > question_count for {rec['package']}"
                )