from datetime import date, datetime
from unittest.mock import Mock

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, inspect, text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

import auth
import main
from database import init_db
from models import Base, Company, PriceSnapshot, AnnualData, User, ReturnAudit, ReturnDocument, ReturnEvent, ReturnPolicy


@pytest.fixture
def api():
    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
    Base.metadata.create_all(engine)
    factory = sessionmaker(bind=engine)
    db = factory()
    year = date.today().year
    db.add_all([Company(ticker="005930", name="테스트", market="KOSPI"),
                User(id=1, google_id="admin", email="admin@example.com", is_admin=True),
                PriceSnapshot(ticker="005930", current_price=10000, market_cap=10000, updated_at=datetime.utcnow()),
                AnnualData(ticker="005930", year=year, is_estimate=True, dividend_per_share=400, net_income=1000, shares_outstanding=100000000),
                AnnualData(ticker="005930", year=year+1, is_estimate=True, dividend_per_share=500),
                AnnualData(ticker="005930", year=year-1, is_estimate=False, shares_outstanding=100000000)])
    db.commit()
    def get_db():
        yield db
    main.app.dependency_overrides[main.get_db] = get_db
    auth_user = {"value": db.get(User, 1)}
    main.app.dependency_overrides[auth.get_current_user] = lambda: auth_user["value"]
    main.invalidate_screener_cache()
    client = TestClient(main.app)
    yield client, db, auth_user, year
    main.app.dependency_overrides.clear()
    main.invalidate_screener_cache()
    db.close()
    engine.dispose()


def payload(year, **kwargs):
    return {"ticker": "005930", "source_key": "manual:first", "source_url": "https://example.com/ir", "title": "소각 결정", "evidence": "보통주 소각 300억원", "status": "pending",
            "events": [{"year": year, "amount": 300, "scope": "common", "evidence": "보통주 소각 300억원"}], **kwargs}


def test_approval_invalidates_cache_and_detail_matches(api):
    client, db, _, year = api
    assert client.get('/api/screener').json()[0]['est_total_shareholder_yield'] is None
    d = client.post('/api/admin/returns/documents', json=payload(year)).json()
    assert d['status'] == 'pending'
    assert client.get('/api/screener').json()[0]['return_status'] == 'unconfirmed'
    response = client.put(f"/api/admin/returns/documents/{d['id']}", json=payload(year, status='approved'))
    assert response.status_code == 200, response.text
    r = client.get('/api/screener').json()[0]
    assert r['est_total_shareholder_yield'] == 7
    detail = client.get(f'/api/stock/005930?year={year}').json()
    assert detail['shareholder_returns']['est_total_shareholder_yield'] == 7
    assert 'history' in detail and 'consensus' in detail
    assert client.get(f'/api/screener?year={year+1}').json()[0]['est_total_shareholder_yield'] is None
    assert client.get(f'/api/stock/005930?year={year-1}').status_code == 400
    assert db.query(ReturnAudit).count() == 2
    assert client.get(f"/api/admin/returns/documents/{d['id']}/audit").status_code == 200


def test_admin_permissions(api):
    client, _, user, year = api
    user['value'] = None
    assert client.get('/api/admin/returns/documents').status_code == 401
    assert client.post('/api/admin/returns/documents', json=payload(year)).status_code == 401
    user['value'] = User(id=2, is_admin=False)
    assert client.get('/api/admin/returns/sources').status_code == 403
    assert client.post('/api/admin/returns/documents', json=payload(year)).status_code == 403
    assert client.get('/api/screener').status_code == 200


def test_automatic_application_cancel_restore_and_audit(api):
    client,db,user,year=api
    d=client.post('/api/admin/returns/documents',json=payload(year,status='active')).json()
    assert d['status'] == 'active'
    assert client.get('/api/screener').json()[0]['est_total_shareholder_yield'] == 7
    path=f"/api/admin/returns/documents/{d['id']}/status"
    assert client.post(path,json={'action':'cancel','reason':'중복 공시'}).json()['status'] == 'excluded'
    assert client.get('/api/screener').json()[0]['est_total_shareholder_yield'] is None
    detail=client.get(f'/api/stock/005930?year={year}').json()['shareholder_returns']
    assert detail['return_sources'][0]['status'] == 'excluded'
    assert detail['return_sources'][0]['events'][0]['amount'] == 300
    assert client.post(path,json={'action':'cancel'}).status_code == 200
    assert db.query(ReturnAudit).count() == 2
    assert client.post(path,json={'action':'restore'}).json()['status'] == 'active'
    assert client.get('/api/screener').json()[0]['est_total_shareholder_yield'] == 7
    audits=client.get(f"/api/admin/returns/documents/{d['id']}/audit").json()
    assert [a['action'] for a in audits] == ['review','cancel','restore']
    assert '중복 공시' in audits[1]['snapshot']
    user['value']=None
    assert client.post(path,json={'action':'cancel'}).status_code == 401
    user['value']=User(id=2,is_admin=False)
    assert client.post(path,json={'action':'cancel'}).status_code == 403


def test_restore_reference_does_not_invent_missing_amount(api):
    client,db,user,year=api
    d=ReturnDocument(ticker='005930',source_key='ref',source_url='https://example.com',title='참고',status='excluded',error='dart_status_014')
    db.add(d);db.commit()
    response=client.post(f'/api/admin/returns/documents/{d.id}/status',json={'action':'restore'})
    assert response.json()['status'] == 'reference'
    assert client.get('/api/screener').json()[0]['est_total_shareholder_yield'] is None


def test_review_list_uses_three_queries_for_many_documents(api):
    from sqlalchemy import event
    client,db,users,year=api
    for i in range(200):
        doc=ReturnDocument(ticker='005930',source_key=f'bulk:{i}',source_url='https://example.com/ir',title=f'소각 {i}',status='pending')
        db.add(doc);db.flush()
        db.add(ReturnEvent(document_id=doc.id,year=year,amount=i,scope='common'))
        db.add(ReturnPolicy(document_id=doc.id,year=year,kind='cancellation',amount=i,scope='common'))
    db.commit()
    assert users['value'].is_admin  # Load authentication state before measuring list queries.
    queries=[]
    def record(conn,cursor,statement,parameters,context,executemany):
        if statement.lstrip().upper().startswith('SELECT'):
            queries.append(statement)
    engine=db.get_bind()
    event.listen(engine,'before_cursor_execute',record)
    try:
        response=client.get('/api/admin/returns/documents')
    finally:
        event.remove(engine,'before_cursor_execute',record)
    assert response.status_code == 200
    documents=response.json()
    assert len(documents) == 200
    assert all(len(d['events']) == len(d['policies']) == 1 for d in documents)
    assert len(queries) == 3


def test_supersession_cycle_and_duplicate_keys(api):
    client, _, _, year = api
    one = client.post('/api/admin/returns/documents', json=payload(year, status='approved')).json()
    assert client.post('/api/admin/returns/documents', json=payload(year)).status_code == 409
    two = client.post('/api/admin/returns/documents', json=payload(year, source_key='manual:second', supersedes_id=one['id'], status='approved')).json()
    assert client.put(f"/api/admin/returns/documents/{one['id']}", json=payload(year, supersedes_id=two['id'], status='approved')).status_code == 422
    assert client.get('/api/screener').json()[0]['est_cancellation_amount'] == 300
    assert client.post('/api/admin/returns/documents', json=payload(year, source_key='manual:third', supersedes_id=one['id'], status='approved')).status_code == 409


def test_correction_requires_previous_source(api):
    client,_,_,year=api
    assert client.post('/api/admin/returns/documents',json=payload(year,title='기재정정 소각 결정',status='approved')).status_code == 422


def test_external_review_changes_invalidate_worker_cache(api):
    from return_api import ReviewInput, write_review
    client,db,_,year=api
    assert client.get('/api/screener').json()[0]['est_cancellation_amount'] is None
    # Simulate another process updating the database without clearing Python caches.
    write_review(db,ReviewInput(**payload(year,status='approved')))
    assert client.get('/api/screener').json()[0]['est_cancellation_amount'] == 300


def test_sources_upsert_and_url_validation(api):
    client, db, _, year = api
    body = {'ticker':'005930','url':'https://example.com/ir','is_index':True}
    one = client.post('/api/admin/returns/sources',json=body).json()
    two = client.post('/api/admin/returns/sources',json={**body,'enabled':False}).json()
    assert one['id'] == two['id'] and not two['enabled']
    assert client.post('/api/admin/returns/sources',json={**body,'url':'file:///etc/passwd'}).status_code == 422


def test_price_refresh_clears_cap_and_invalidates_cache(api, monkeypatch):
    client, db, _, year = api
    client.post('/api/admin/returns/documents',json=payload(year,status='approved'))
    assert client.get('/api/screener').json()[0]['est_total_shareholder_yield'] == 7
    import requests
    monkeypatch.setattr(requests, 'get', lambda *a, **k: Mock(raise_for_status=lambda:None, json=lambda:{'closePrice':'20,000'}))
    response = client.post('/api/refresh/price/005930')
    assert response.json()['market_cap'] == 20000
    assert client.get('/api/screener').json()[0]['est_total_shareholder_yield'] == 3.5
    old = db.query(AnnualData).filter_by(is_estimate=False).first()
    old.shares_outstanding = None
    db.commit()
    assert client.post('/api/refresh/price/005930').json()['market_cap'] is None
    assert client.get('/api/screener').json()[0]['est_total_shareholder_yield'] is None


def test_existing_sqlite_database_upgrade_preserves_data(monkeypatch):
    import database
    engine = create_engine('sqlite://', poolclass=StaticPool)
    # Simulate the pre-feature schema and a populated company record.
    old_tables = [t for name,t in Base.metadata.tables.items() if not name.startswith('return_')]
    Base.metadata.create_all(engine, tables=old_tables)
    with engine.begin() as conn:
        conn.execute(text("INSERT INTO company (ticker,name) VALUES ('005930','기존기업')"))
    monkeypatch.setattr(database,'engine',engine)
    monkeypatch.setattr(database,'_is_postgres',False)
    init_db()
    init_db()
    assert 'return_document' in inspect(engine).get_table_names()
    with engine.connect() as conn:
        assert conn.execute(text("SELECT name FROM company WHERE ticker='005930'")).scalar() == '기존기업'
    engine.dispose()


def test_postgres_schema_compiles_without_altering_existing_tables():
    from sqlalchemy.schema import CreateTable
    from sqlalchemy.dialects import postgresql
    for name,table in Base.metadata.tables.items():
        if name.startswith('return_'):
            sql = str(CreateTable(table).compile(dialect=postgresql.dialect()))
            assert 'CREATE TABLE' in sql