"""Independently expected operational totals on invented tenant records only."""
import csv
import io
import json
import sqlite3

import pytest
from fastapi.testclient import TestClient
from backend import reporting
from backend.reporting import Filters, ExportFilters, build_report, export_csv
from test_boundaries import app, client, write, job, quote, invoice, portal, ORIGIN
from test_documents import act, doc, schedule
from test_workflow import edit
from test_workforce import person, entry
from test_materials import material
from test_customer_reviews import customer, selection, preview, confirm
from test_recovery import table_rows
from test_identity import add_staff

URL = '/api/reports/activity'
RANGE = dict(start='2026-10-01', end='2026-10-31')
OBSERVED = '2026-10-28T18:00:00+00:00'


def report(c, **filters):
    r = c.get(URL, params={**RANGE, **filters})
    assert r.status_code == 200, r.text
    return r.json()


def saved_job(c, service='Joinery', name='Morgan Dêmø', email='shared@fictional.example'):
    r = write(c, '/api/jobs', dict(name=name, email=email, phone='000000', address='Invented address', service=service, message='Fictional work only'))
    assert r.status_code == 200, r.text
    return r.json()


def receipt(c, i, amount, paid_on='2026-10-12'):
    r = write(c, '/api/invoices/' + i['id'] + '/payments', dict(amount_pence=amount, paid_on=paid_on, note='PRIVATE PAYMENT NOTE'))
    assert r.status_code == 200, r.text
    return r.json()


def csv_rows(response):
    assert response.status_code == 200, response.text
    assert response.content.startswith(b'\xef\xbb\xbf')
    return list(csv.DictReader(io.StringIO(response.content.decode('utf-8-sig'))))


def export(c, data, **choice):
    return c.get(URL + '/export', params={**data['filters'], 'snapshot': data['snapshot'], **choice})


@pytest.fixture
def calculation(app, monkeypatch):
    # Times are explicit fixture facts. No production history is edited.
    monkeypatch.setattr(reporting, 'now', lambda: OBSERVED)
    import backend.app as app_module
    import backend.documents as docs
    monkeypatch.setattr(app_module, 'now', lambda: '2026-10-03T12:00:00+00:00')
    monkeypatch.setattr(docs, 'now', lambda: '2026-10-06T12:00:00+00:00')
    c = client(app)
    a = saved_job(c)
    b = saved_job(c, 'Repairs')  # identical contact text, a different exact customer
    old = saved_job(c, 'Repairs', 'Older Customer')
    target = saved_job(c, name='Retained Customer', email='retained@fictional.example')
    with app.state.store.tenant('oak') as db:
        db.execute("UPDATE jobs SET created_at='2026-09-30T12:00:00+00:00' WHERE id=?", (old['id'],))
    edit(c, a, tags=['repeat', 'joinery'], status='completed', completed_date='2026-10-15')
    edit(c, b, tags=['repeat'], status='cancelled')
    q0 = quote(c, a, 10000)
    assert act(c, 'quote', q0, 'share').status_code == 200
    q = act(c, 'quote', q0, 'revise').json()
    from test_documents import edit_payload
    assert write(c, '/api/quotes/' + q['id'], edit_payload(doc(c, 'quote', q), 15000), method='put').status_code == 200
    assert act(c, 'quote', q, 'share').status_code == 200
    assert act(c, 'quote', q, 'accept').status_code == 200
    i = act(c, 'quote', q, 'invoice').json()
    stages = schedule(c, i, (5000, 10000))
    assert act(c, 'invoice', i, 'issue').status_code == 200
    assert write(c, '/api/invoices/' + i['id'] + '/schedule/' + stages[0]['id'] + '/request', dict(version=doc(c, 'invoice', i)['version'])).status_code == 200
    wrong = receipt(c, i, 6000)
    corrected = write(c, '/api/invoices/' + i['id'] + '/payments/' + wrong['id'] + '/correct',
                      dict(version=doc(c, 'invoice', i)['version'], recording_error=True, reason='Correct invented recording error', replacement=dict(amount_pence=4000, paid_on='2026-10-12', note='PRIVATE REPLACEMENT NOTE')))
    assert corrected.status_code == 200, corrected.text
    receipt(c, i, 1000, '2026-09-30')
    qb = quote(c, b, 8000)
    ib = invoice(c, qb)
    assert act(c, 'invoice', ib, 'void').status_code == 200
    edit(c, b, archived=True)
    cb = customer(c, b['customer_id'])
    assert write(c, '/api/customers/' + cb['id'] + '/record-state', dict(version=cb['version'], state='archived', reason='Retain invented closed record', confirmed=True)).status_code == 200
    qo = quote(c, old, 9000)
    assert act(c, 'quote', qo, 'share').status_code == 200
    qr = act(c, 'quote', qo, 'revise').json()
    assert write(c, '/api/quotes/' + qr['id'], edit_payload(doc(c, 'quote', qr), 12000), method='put').status_code == 200
    assert act(c, 'quote', qr, 'share').status_code == 200
    expired = quote(c, old, 4000)
    payload = edit_payload(doc(c, 'quote', expired)); payload['valid_until'] = '2020-01-01'
    assert write(c, '/api/quotes/' + expired['id'], payload, method='put').status_code == 200
    assert act(c, 'quote', expired, 'share').status_code == 409  # an expired draft cannot be shared
    accepted = quote(c, old, 7000)
    assert act(c, 'quote', accepted, 'accept').status_code == 200
    draft_inv = act(c, 'quote', accepted, 'invoice').json()
    p = person(c, name='Taylor Original')
    entry(c, a, p, work_date='2026-10-15', duration_minutes=90)
    entry(c, a, p, work_date='2026-10-15', duration_minutes=1, hourly_cost_override='0.30')
    archived = entry(c, b, p, work_date='2026-10-10', duration_minutes=60, hourly_cost_override='10.00')
    assert write(c, '/api/time/' + archived['id'] + '/archive', dict(version=archived['version'])).status_code == 200
    assert write(c, '/api/team/' + p['id'], dict(version=1, name='Taylor Current', role='Installer', hourly_cost='25.00', colour='#245648'), method='put').status_code == 200
    assert write(c, '/api/team/' + p['id'] + '/archive', dict(version=2)).status_code == 200
    material(c, a, quantity='2', cost_price='12.34', status='received', received_date='2026-10-15')
    retired = material(c, b, quantity='1', cost_price='7.50', status='received', received_date='2026-10-10')
    assert write(c, '/api/materials/' + retired['id'] + '/archive', dict(version=1)).status_code == 200
    material(c, old, cost_price='10.00', status='ordered', ordered_date='2026-10-10')
    plan = preview(c, selection(c, a, target))
    assert plan['can_confirm'], plan
    assert confirm(c, plan).status_code == 200
    return dict(c=c, a=a, b=b, old=old, target=target, q=q, i=i, ib=ib, p=p, draft_inv=draft_inv)


def test_exact_totals_revisions_stages_reversals_archive_and_regrouping(calculation, app):
    x = calculation; data = report(x['c']); total = data['totals']
    # £150 issued; £60 erroneous receipt reversed/replaced by £40, plus £10 in September.
    # £80 void, £70 draft, £50/£100 stage requests, prior revisions and document rows add nothing.
    assert total['jobs_created'] == 3
    assert total['completed_jobs'] == 1
    assert total['quote_revisions_created'] == 7
    assert total['issued_invoice_value_pence'] == 15000
    assert total['recorded_receipts_pence'] == 4000
    assert total['issued_invoice_balance_pence'] == 10000
    assert total['void_invoice_value_pence'] == 8000
    assert total['reversed_receipt_pence'] == 6000
    assert total['receipt_reversals_recorded'] == 1
    assert total['open_quote_estimate_pence'] == 12000
    # 90 minutes at £20/hour + 1 minute at 30p/hour = £30.01 (half-up per entry).
    assert (total['person_minutes'], total['manual_person_minutes'], total['field_person_minutes']) == (91, 91, 0)
    assert total['internal_labour_cost_pence'] == 3001
    assert total['archived_person_minutes'] == 60 and total['archived_labour_cost_pence'] == 1000
    assert total['received_material_cost_pence'] == 2468 and total['archived_material_cost_pence'] == 750
    customer_groups = {g['id']: g for g in data['groups']['customer']}
    assert x['a']['customer_id'] not in customer_groups
    assert customer_groups[x['target']['customer_id']]['totals']['issued_invoice_value_pence'] == 15000
    assert customer_groups[x['b']['customer_id']]['state'] == 'archived'
    record = next(r for r in data['records'] if r['metric'] == 'issued_invoice_value_pence')
    assert record['customer_id'] == x['target']['customer_id'] and record['original_customer_id'] == x['a']['customer_id']
    team = data['groups']['team'][0]
    assert team['label'] == 'Taylor Current' and team['state'] == 'archived'
    assert team['worked_job_ids'] == [x['a']['id']]
    assert all('invoice' not in k and 'receipt' not in k for k in team['totals'])
    times = [r for r in data['records'] if r['metric'] == 'person_minutes']
    assert all(r['resource_name'] == 'Taylor Original' for r in times)
    operations = x['c'].get('/api/reports/operations', params=RANGE).json()
    for key in operations['period_money']:
        assert operations['period_money'][key] == total[key]


def test_period_filters_group_drills_and_same_name_customers(calculation):
    x = calculation; c = x['c']; full = report(c)
    for dimension, field in (('service', 'service_key'), ('customer', 'customer_id'), ('team', 'resource_id')):
        for group in full['groups'][dimension]:
            for key, total in group['totals'].items():
                assert sum(r['value'] for r in full['records'] if r['metric'] == key and r.get(field) == group['id']) == total
    assert report(c, customer_id=x['a']['customer_id'])['matched_jobs'] == 0  # alias not silently followed
    assert report(c, customer_id=x['target']['customer_id'])['matched_jobs'] == 2
    archived = report(c, archive='archived')
    assert archived['matched_jobs'] == 1 and archived['totals']['void_invoice_value_pence'] == 8000
    assert report(c, tag='repeat', stage='completed')['matched_jobs'] == 1
    joinery = report(c, service=reporting.service_key('Joinery'))
    assert joinery['totals']['internal_labour_cost_pence'] == 3001
    september = report(c, start='2026-09-01', end='2026-09-30')
    assert september['totals']['recorded_receipts_pence'] == 1000 and september['totals']['issued_invoice_value_pence'] == 0
    assert september['totals']['issued_invoice_balance_pence'] == full['totals']['issued_invoice_balance_pence']
    assert september['totals']['open_quote_estimate_pence'] == full['totals']['open_quote_estimate_pence']
    assert len(full['options']['customers']) == 4  # no name/email grouping


def test_nested_cohort_and_tag_stage_distributions_are_one_population(calculation):
    data = report(calculation['c'])
    # The void invoice's job retains its shared/accepted quote evidence, but
    # does not have a currently issued invoice. The zero-document target is the third job.
    assert [r['count'] for r in data['funnel']] == [3, 2, 2, 1, 1]
    assert [r['percent_tenths'] for r in data['funnel']] == [1000, 667, 667, 333, 333]
    assert all(r['population'] == 3 for r in data['funnel'])
    assert sum(map(len, data['cohort_stages'].values())) == 3
    assert len(data['cohort_tags']['repeat']) == 2 and len(data['cohort_tags']['joinery']) == 1
    assert data['totals']['quote_revisions_created'] > data['totals']['cohort_shared']


@pytest.mark.parametrize('changes', [dict(start='20261001'), dict(end='2026-02-30'), dict(start='2026-11-01'), dict(start='2024-01-01'), dict(customer_id='foreign'), dict(stage='invented'), dict(archive='deleted'), dict(unrecognised='ignored')])
def test_invalid_filters_fail_explicitly_without_writes(app, changes):
    c = client(app); before = table_rows(app.state.store.root, 'oak')
    assert c.get(URL, params={**RANGE, **changes}).status_code == 422
    assert table_rows(app.state.store.root, 'oak') == before


def test_repeated_filters_and_foreign_ids_never_fall_back_to_all(app):
    c = client(app); other = client(app, 1); foreign = job(other); resource = person(other)
    assert c.get(URL + '?start=2026-10-01&start=2026-10-02&end=2026-10-31').status_code == 422
    for params in (dict(customer_id=foreign['customer_id']), dict(service='f' * 64), dict(tag='foreign')):
        assert c.get(URL, params={**RANGE, **params}).status_code == 404
    empty = report(c)
    for dimension, gid in (('customer', foreign['customer_id']), ('team', resource['id']), ('service', 'f' * 64)):
        assert export(c, empty, dimension=dimension, group_id=gid).status_code == 404
    assert empty['matched_jobs'] == 0 and empty['records'] == []
    assert all(v == 0 for v in empty['totals'].values()) and all(v['percent_tenths'] is None for v in empty['funnel'])
    assert csv_rows(export(c, empty, kind='records', metric='person_minutes'))[0]['row_type'] == 'empty'


def test_owner_only_and_no_private_bodies_no_mutation(calculation, app):
    x = calculation; c = x['c']; staff, _ = add_staff(app, c); p, _ = portal(app, c, x['target'])
    before = table_rows(app.state.store.root, 'oak'); data = report(c)
    raw = json.dumps(data) + export(c, data).text
    for text in ('PRIVATE PAYMENT NOTE', 'PRIVATE REPLACEMENT NOTE', 'Private margin note', 'shared@fictional.example', 'portal_hash', 'password', 'csrf'):
        assert text not in raw
    for other, expected in ((staff, 403), (p, 401), (TestClient(app, headers=ORIGIN), 401)):
        assert other.get(URL, params=RANGE).status_code == expected
        assert export(other, data).status_code == expected
    assert table_rows(app.state.store.root, 'oak') == before
    assert report(client(app, 1))['matched_jobs'] == 0
    assert c.get(URL, params=RANGE, headers={'x-business-id':'harbour'}).status_code == 409
    assert c.get(URL, params=RANGE).headers['cache-control'] == 'no-store'


def test_export_matches_exact_screen_and_stale_changes_reject(calculation):
    x = calculation; c = x['c']; data = report(c)
    rows = csv_rows(export(c, data, dimension='service'))
    assert sum(int(r['value']) for r in rows if r['metric'] == 'issued_invoice_value_pence') == 15000
    records = csv_rows(export(c, data, kind='records', metric='recorded_receipts_pence', dimension='customer', group_id=x['target']['customer_id']))
    assert len(records) == 1 and records[0]['value'] == '4000'
    assert records[0]['original_customer_id'] == x['a']['customer_id']
    assert records[0]['generated_at_utc'] == OBSERVED and records[0]['period_end_uk'] == '2026-10-31'
    assert records[0]['scope'] == 'period' and records[0]['unit'] == 'pence'
    assert 'bank-cleared' in records[0]['definition'] and json.loads(records[0]['filters_json']) == data['filters']
    assert export(c, data, dimension='team', metric='recorded_receipts_pence').status_code == 422
    edit(c, x['target'], tags=['changed'])
    assert export(c, data).status_code == 409
    assert export(c, report(c)).status_code == 200


@pytest.mark.parametrize('name', ['=1+2', '+SUM(A1)', '-1+2', '@formula', '\ufeff=2', '\u200b＋2', '\t=2'])
def test_csv_text_formula_unicode_quotes_and_newlines(app, name):
    c = client(app); j = saved_job(c, service='Oak, "quoted"\nDêmø', name=name)
    data = report(c); rows = csv_rows(export(c, data, kind='records', metric='open_jobs'))
    assert rows[0]['service_label'] == 'Oak, "quoted"\nDêmø'
    assert rows[0]['customer_name'].startswith("'")
    assert rows[0]['job_id'] == j['id'] and rows[0]['value'] == '1'


@pytest.mark.parametrize('stamp,expected', [
    ('2026-09-30T22:59:59+00:00', '2026-09-30'), ('2026-09-30T23:00:00+00:00', '2026-10-01'),
    ('2026-10-24T23:30:00+00:00', '2026-10-25'), ('2026-10-25T01:30:00+00:00', '2026-10-25'),
    ('2026-03-28T23:59:59+00:00', '2026-03-28'), ('2026-03-29T23:00:00+01:00', '2026-03-29'),
    ('2026-10-01', None), ('not-a-date', None), ('2026-10-01T12:00:00', None), (None, None)])
def test_uk_timestamp_boundary_and_unknown(stamp, expected):
    assert reporting.uk_day(stamp) == expected


def test_date_only_values_do_not_shift_and_unknown_dates_stay_unknown(app, monkeypatch):
    c = client(app); j = job(c); edit(c, j, status='completed'); edit(c, j, completed_date='')
    with app.state.store.tenant('oak') as db:
        db.execute("UPDATE jobs SET created_at='2026-09-30T23:00:00+00:00'")
        db.execute("UPDATE jobs SET service=''")
    data = report(c, start='2026-10-01', end='2026-10-01')
    assert data['totals']['jobs_created'] == 1 and data['totals']['completed_jobs'] == 0
    assert data['totals']['unknown_dates'] == 1 and data['groups']['service'][0]['label'] == 'Unclassified'
    assert data['records'][next(n for n, r in enumerate(data['records']) if r['metric'] == 'unknown_dates')]['date_role'] == 'saved completion'
    assert reporting.day_only('2026-10-01') == '2026-10-01'
    with app.state.store.tenant('oak') as db:
        db.execute("UPDATE jobs SET created_at='unknown'")
    unknown = report(c)
    assert unknown['totals']['cohort_created'] == 0 and unknown['totals']['unknown_dates'] == 2


@pytest.mark.parametrize('bound,value', [('MAX_INPUT', 0), ('MAX_JOBS', 0), ('MAX_FACTS', 0), ('MAX_EXPORT', 0)])
def test_oversized_requests_fail_without_partial_totals_or_file(app, monkeypatch, bound, value):
    c = client(app); job(c); data = report(c); before = table_rows(app.state.store.root, 'oak')
    monkeypatch.setattr(reporting, bound, value)
    response = export(c, data) if bound == 'MAX_EXPORT' else c.get(URL, params=RANGE)
    # Limits are part of the snapshot. Recompute only for the export-bound fixture.
    if bound == 'MAX_EXPORT':
        response = export(c, report(c))
    assert response.status_code == 422 and 'partial' in response.text
    assert table_rows(app.state.store.root, 'oak') == before


def test_query_only_snapshot_rejects_an_accidental_write(app, monkeypatch):
    c = client(app); job(c); before = table_rows(app.state.store.root, 'oak')
    def attempted_write(db, filters):
        db.execute("UPDATE jobs SET status='cancelled'")
    monkeypatch.setattr(reporting, 'build_report', attempted_write)
    with pytest.raises(sqlite3.OperationalError, match='readonly'):
        c.get(URL, params=RANGE)
    assert table_rows(app.state.store.root, 'oak') == before


def test_field_conversion_generations_materials_and_document_sources_count_once(app):
    from test_fieldwork import captured, review, convert, timer, base, draft, start, act as timer_action
    from datetime import datetime, timedelta
    c = client(app); j, people, clock, tid = captured(c, app, team_size=2)
    # 90 seconds rounds to 2 minutes for each person, then a reviewed correction
    # produces one active minute each and retains the old entries as archived.
    interval = timer(c, j, tid)['raw_intervals'][0]
    interval['end'] = (datetime.fromisoformat(interval['start']) + timedelta(seconds=60)).isoformat()
    assert review(c, j, tid, [interval]).status_code == 200
    assert convert(c, j, tid).status_code == 200
    entry(c, j, people[0], work_date='2026-10-15', duration_minutes=30)
    log = write(c, base(j) + '/logs', dict(kind='materials', content='PRIVATE FIELD MATERIAL BODY', quantity='1.125', unit='box', work_date='2026-10-25', photo_ids=[])).json()
    assert write(c, base(j) + '/logs/' + log['id'] + '/state/review', dict(version=1, reason='Check invented material')).status_code == 200
    assert write(c, base(j) + '/logs/' + log['id'] + '/material', dict(version=2, reason='Review received costs', cost_price='1.23', status='received')).status_code == 200
    assert write(c, base(j) + '/logs', dict(kind='note', content='PRIVATE UNREVIEWED BODY', work_date='2026-10-25')).status_code == 200
    pending = start(c, j, people).json()['id']; clock.advance(60)
    assert timer_action(c, j, pending, 'stop').status_code == 200
    source_invoice = draft(c, j).json()
    assert act(c, 'invoice', source_invoice, 'issue').status_code == 200
    before = table_rows(app.state.store.root, 'oak'); data = report(c)
    assert data['totals']['person_minutes'] == 32  # 30 manual + 1 each for two people
    assert data['totals']['field_person_minutes'] == 2
    assert data['totals']['internal_labour_cost_pence'] == 1066  # £10 + 33p + 33p
    assert data['totals']['archived_person_minutes'] == 4
    assert data['totals']['archived_labour_cost_pence'] == 134
    assert data['totals']['received_material_cost_pence'] == 138
    assert data['totals']['field_logs_reviewed'] == data['totals']['field_logs_unreviewed'] == 1
    assert data['totals']['field_sessions_unconverted'] == 1
    assert data['totals']['cohort_direct_invoice'] == 1
    assert data['totals']['cohort_issued'] == 0  # direct source-backed invoice is a separate path
    assert len(data['groups']['team']) == 2
    assert 'PRIVATE FIELD' not in json.dumps(data) and 'PRIVATE UNREVIEWED' not in json.dumps(data)
    assert table_rows(app.state.store.root, 'oak') == before


def test_selected_job_reassignment_keeps_other_jobs_and_original_document(calculation):
    x = calculation; c = x['c']
    source = saved_job(c, name='Identical Person'); separate = saved_job(c, name='Identical Person')
    other = write(c, '/api/customers/' + source['customer_id'] + '/jobs', dict(customer_version=customer(c, source['customer_id'])['version'], service='Other work', message='Invented second job')).json()
    q = quote(c, source, 12345); i = invoice(c, q)
    before = report(c)['totals']['issued_invoice_value_pence']
    plan = preview(c, selection(c, source, separate, kind='jobs', job_ids=[source['id']]))
    assert confirm(c, plan).status_code == 200
    assert report(c)['totals']['issued_invoice_value_pence'] == before
    source_report = report(c, customer_id=source['customer_id']); target = report(c, customer_id=separate['customer_id'])
    assert source_report['matched_jobs'] == 1 and source_report['totals']['issued_invoice_value_pence'] == 0
    assert target['matched_jobs'] == 2 and target['totals']['issued_invoice_value_pence'] == 12345
    r = next(r for r in target['records'] if r['metric'] == 'issued_invoice_value_pence')
    assert r['original_customer_id'] == source['customer_id'] and r['customer_id'] == separate['customer_id']


def test_cohort_later_evidence_and_observation_point_are_explicit(app):
    c = client(app); j = job(c); q = quote(c, j); i = invoice(c, q); receipt(c, i, 100)
    with app.state.store.tenant('oak') as db:
        db.execute("UPDATE jobs SET created_at='2026-09-03T00:00:00+00:00'")
        db.execute("UPDATE quote_links SET shared_at='2026-10-02T12:00:00+00:00',accepted_at='2026-10-03T12:00:00+00:00'")
        db.execute("UPDATE invoices SET issued_at='2026-10-04T12:00:00+00:00'")
        db.execute("UPDATE payments SET created_at='2026-10-12T12:00:00+00:00'")
    with app.state.store.tenant('oak') as db:
        late = build_report(db, Filters(start='2026-09-01', end='2026-09-30'), generated_at=OBSERVED)
        early = build_report(db, Filters(start='2026-09-01', end='2026-09-30'), generated_at='2026-10-02T13:00:00+00:00')
    assert [x['count'] for x in late['funnel']] == [1, 1, 1, 1, 1]
    assert late['totals']['issued_invoice_value_pence'] == late['totals']['recorded_receipts_pence'] == 0
    assert [x['count'] for x in early['funnel']] == [1, 1, 0, 0, 0]


def test_export_is_byte_deterministic_and_date_range_bounds_are_inclusive(calculation, app):
    c = calculation['c']; data = report(c)
    assert export(c, data, kind='summary', dimension='customer').content == export(c, data, kind='summary', dimension='customer').content
    assert c.get(URL, params=dict(start='2024-01-01', end='2024-12-31')).status_code == 200
    assert c.get(URL, params=dict(start='2024-01-01', end='2025-01-01')).status_code == 422
    assert report(c, start='2026-10-12', end='2026-10-12')['totals']['recorded_receipts_pence'] == 4000
    assert report(c, start='2026-10-13', end='2026-10-13')['totals']['recorded_receipts_pence'] == 0


def test_published_definition_registry_matches_calculations():
    from pathlib import Path
    registry = Path(__file__).resolve().parents[1] / 'docs/reporting-metrics.json'
    assert json.loads(registry.read_text()) == reporting.METRICS