CIAgent/apps/api/tests/integration/test_change_detection_service.py

"""End-to-end (DB-backed) tests for change_detection_service: two real
Snapshot rows in, a DetectedChange (or None) out. Collectors themselves are
already covered elsewhere; this exercises the diff/scoring/persistence
pipeline directly against SQLite."""

from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select
from sqlalchemy.orm import selectinload

from app.models.company import Company
from app.models.detected_change import DetectedChange
from app.models.enums import ChangeStatus, ChangeType, SourceType
from app.models.snapshot import Snapshot
from app.models.source import Source
from app.repositories.source_repository import SnapshotRepository, SourceRepository
from app.services.change_detection_service import detect_change_for_source


async def _make_company(db_session, *, monitoring_focus: str | None = None) -> Company:
    company = Company(
        id=uuid.uuid4(),
        user_id=uuid.uuid4(),
        name="Acme Corp",
        slug=f"acme-corp-{uuid.uuid4().hex[:6]}",
        monitoring_focus=monitoring_focus,
    )
    db_session.add(company)
    await db_session.commit()
    result = await db_session.execute(
        select(Company).where(Company.id == company.id).options(selectinload(Company.aliases))
    )
    return result.scalar_one()


async def _make_source(
    db_session, company: Company, *, source_type: SourceType = SourceType.WEBSITE, trust_score=0.8
) -> Source:
    return await SourceRepository(db_session).create(
        company_id=company.id,
        source_type=source_type,
        name="Test Source",
        base_url="https://example.com",
        trust_score=trust_score,
    )


async def _make_snapshot(
    db_session,
    company: Company,
    source: Source,
    *,
    content_hash: str,
    urls: list[str],
    text_summary: str,
    created_at: datetime | None = None,
) -> Snapshot:
    snapshot = await SnapshotRepository(db_session).create(
        company_id=company.id,
        source_id=source.id,
        snapshot_type=source.source_type.value,
        hash=content_hash,
        structured_summary={"urls": urls, "titles": [], "document_count": len(urls)},
        text_summary=text_summary,
    )
    if created_at is not None:
        snapshot.created_at = created_at
        await db_session.commit()
    return snapshot


@pytest.mark.asyncio
async def test_no_previous_snapshot_produces_no_change(db_session):
    company = await _make_company(db_session)
    source = await _make_source(db_session, company)
    current = await _make_snapshot(
        db_session, company, source, content_hash="h1", urls=["/a"], text_summary="A"
    )

    result = await detect_change_for_source(db_session, source, company, current, uuid.uuid4())
    assert result is None


@pytest.mark.asyncio
async def test_identical_hash_produces_no_change(db_session):
    company = await _make_company(db_session)
    source = await _make_source(db_session, company)
    await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="same-hash",
        urls=["/a"],
        text_summary="A",
        created_at=datetime.now(UTC) - timedelta(hours=1),
    )
    current = await _make_snapshot(
        db_session, company, source, content_hash="same-hash", urls=["/a"], text_summary="A"
    )

    result = await detect_change_for_source(db_session, source, company, current, uuid.uuid4())
    assert result is None


@pytest.mark.asyncio
async def test_new_item_detected_as_new_document(db_session):
    company = await _make_company(db_session)
    source = await _make_source(db_session, company)
    await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h1",
        urls=["/careers/job-a"],
        text_summary="### Job A\nExisting posting.",
        created_at=datetime.now(UTC) - timedelta(hours=1),
    )
    current = await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h2",
        urls=["/careers/job-a", "/careers/job-b"],
        text_summary="### Job A\nExisting posting.\n\n### Job B\nNew battery engineer role.",
    )

    change = await detect_change_for_source(db_session, source, company, current, uuid.uuid4())

    assert change is not None
    assert change.change_type == ChangeType.NEW_DOCUMENT
    assert change.status == ChangeStatus.NEW
    assert "/careers/job-b" in change.raw_diff["structured_added"]


@pytest.mark.asyncio
async def test_leadership_mention_takes_priority_over_new_document(db_session):
    company = await _make_company(db_session)
    source = await _make_source(db_session, company)
    await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h1",
        urls=["/press/1"],
        text_summary="### Old release\nRoutine update.",
        created_at=datetime.now(UTC) - timedelta(hours=1),
    )
    current = await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h2",
        urls=["/press/1", "/press/2"],
        text_summary=(
            "### Old release\nRoutine update.\n\n"
            "### New release\nJane Smith has been named the company's new CEO effective immediately."
        ),
    )

    change = await detect_change_for_source(db_session, source, company, current, uuid.uuid4())

    # This test is about classification priority (leadership beats the
    # new-document signal that's also present), not about the resulting
    # severity number - see test_scoring.py for severity-formula coverage.
    assert change is not None
    assert change.change_type == ChangeType.LEADERSHIP_CHANGE


@pytest.mark.asyncio
async def test_price_mention_change_detected(db_session):
    company = await _make_company(db_session)
    source = await _make_source(db_session, company)
    await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h1",
        urls=["/pricing"],
        text_summary="### Pricing\nThe base plan is $49/month.",
        created_at=datetime.now(UTC) - timedelta(hours=1),
    )
    current = await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h2",
        urls=["/pricing"],
        text_summary="### Pricing\nThe base plan is $59/month.",
    )

    change = await detect_change_for_source(db_session, source, company, current, uuid.uuid4())

    assert change is not None
    assert change.change_type == ChangeType.PRICE_CHANGE


@pytest.mark.asyncio
async def test_sec_edgar_new_filing_detected(db_session):
    company = await _make_company(db_session)
    source = await _make_source(db_session, company, source_type=SourceType.SEC_EDGAR)
    await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h1",
        urls=["/filings/10-K-2025"],
        text_summary="### 10-K\nAnnual report.",
        created_at=datetime.now(UTC) - timedelta(hours=1),
    )
    current = await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h2",
        urls=["/filings/10-K-2025", "/filings/8-K-2026"],
        text_summary="### 10-K\nAnnual report.\n\n### 8-K\nCurrent report.",
    )

    change = await detect_change_for_source(db_session, source, company, current, uuid.uuid4())

    assert change is not None
    assert change.change_type == ChangeType.FILING_NEW


@pytest.mark.asyncio
async def test_content_modified_below_threshold_is_not_reported(db_session):
    company = await _make_company(db_session)
    source = await _make_source(db_session, company)
    # Bounded text diff operates line-by-line - many lines so that changing
    # one of them is a small fraction of the whole, not "the whole line
    # differs" (which is what a single long line would produce instead).
    lines = [f"line {i} says something routine about the company" for i in range(50)]
    previous_text = "\n".join(lines)
    await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h1",
        urls=["/about"],
        text_summary=previous_text,
        created_at=datetime.now(UTC) - timedelta(hours=1),
    )
    # Change just one line out of 50 - below the content-modified threshold.
    lines[25] = "line 25 says something slightly different about the company"
    current = await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h2",
        urls=["/about"],
        text_summary="\n".join(lines),
    )

    change = await detect_change_for_source(db_session, source, company, current, uuid.uuid4())
    assert change is None


@pytest.mark.asyncio
async def test_focus_match_is_recorded_via_higher_significance(db_session):
    matching_company = await _make_company(
        db_session, monitoring_focus="Watch for battery manufacturing expansion"
    )
    other_company = await _make_company(db_session, monitoring_focus="Watch for pricing changes")

    matching_source = await _make_source(db_session, matching_company)
    other_source = await _make_source(db_session, other_company)

    old_text = "### Careers\nExisting listing."
    new_text = (
        "### Careers\nExisting listing.\n\n### New role\nBattery manufacturing engineer wanted."
    )

    for company, source in ((matching_company, matching_source), (other_company, other_source)):
        await _make_snapshot(
            db_session,
            company,
            source,
            content_hash="h1",
            urls=["/careers/a"],
            text_summary=old_text,
            created_at=datetime.now(UTC) - timedelta(hours=1),
        )

    matching_current = await _make_snapshot(
        db_session,
        matching_company,
        matching_source,
        content_hash="h2",
        urls=["/careers/a", "/careers/b"],
        text_summary=new_text,
    )
    other_current = await _make_snapshot(
        db_session,
        other_company,
        other_source,
        content_hash="h2",
        urls=["/careers/a", "/careers/b"],
        text_summary=new_text,
    )

    matched_change = await detect_change_for_source(
        db_session, matching_source, matching_company, matching_current, uuid.uuid4()
    )
    unmatched_change = await detect_change_for_source(
        db_session, other_source, other_company, other_current, uuid.uuid4()
    )

    assert matched_change.significance_score > unmatched_change.significance_score


@pytest.mark.asyncio
async def test_exact_repeat_within_cooldown_is_suppressed(db_session):
    company = await _make_company(db_session)
    source = await _make_source(db_session, company)

    await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h1",
        urls=["/a"],
        text_summary="A",
        created_at=datetime.now(UTC) - timedelta(hours=2),
    )
    snap2 = await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h2",
        urls=["/a", "/b"],
        text_summary="A\nB",
        created_at=datetime.now(UTC) - timedelta(hours=1),
    )
    first_change = await detect_change_for_source(db_session, source, company, snap2, uuid.uuid4())
    assert first_change is not None

    # A third snapshot reverts to hash h1's item set momentarily, then a
    # fourth snapshot reproduces the *exact same* added-item diff as before -
    # this should be suppressed as a repeat within the cooldown window.
    await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h1-again",
        urls=["/a"],
        text_summary="A",
    )
    snap4 = await _make_snapshot(
        db_session,
        company,
        source,
        content_hash="h2-again",
        urls=["/a", "/b"],
        text_summary="A\nB",
    )
    second_change = await detect_change_for_source(db_session, source, company, snap4, uuid.uuid4())

    assert second_change is None
    all_changes = (
        (
            await db_session.execute(
                select(DetectedChange).where(DetectedChange.source_id == source.id)
            )
        )
        .scalars()
        .all()
    )
    assert len(all_changes) == 1