"""Behavioral tests for parsing, classification, transactions and export boundaries."""
import contextlib,io,json,sqlite3,tempfile,unittest,threading
from pathlib import Path
from unittest.mock import patch
from fixtures import xml
from report import render
import triage
class TriageTests(unittest.TestCase):
    def setUp(self):self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name);self.db=self.root/'history.sqlite';self.path=self.root/'results.xml'
    def tearDown(self):self.tmp.cleanup()
    def write(self,cases):self.path.write_bytes(xml(cases));return self.path
    def add(self,cases,sequence=1,attempt=1,mode='full',job='linux'):
        self.write(cases);return triage.ingest(self.db,[self.path],build=f'b{sequence}',sequence=sequence,job=job,attempt=attempt,mode=mode)
    def rows(self,build='b1'):return triage.snapshot(self.db)['selections'][build]['rows']
    def test_namespace_nested_and_missing_duration(self):
        row=triage.parse_xml(b'<j:testsuites xmlns:j="urn:j"><j:testsuite><j:testsuite><j:testcase name="a"/></j:testsuite></j:testsuite></j:testsuites>')[0];self.assertEqual(row['status'],'pass');self.assertIsNone(row['duration'])
    def test_reject_dtd_and_utf16(self):
        for data in [b'<!DOCTYPE x [<!ENTITY x "expanded">]><testsuite><testcase name="a"/></testsuite>',xml([{'name':'a'}]).decode().encode('utf-16')]:
            with self.assertRaises(triage.InvalidReport):triage.parse_xml(data)
    def test_invalid_or_empty_xml(self):
        for data in [b'<testsuite>',b'<testsuite/>',b'<other/>',b'<testsuite><testcase/></testsuite>']:
            with self.assertRaises(triage.InvalidReport):triage.parse_xml(data)
    def test_bad_duration(self):
        for value in ['NaN','Infinity',-1,604801,'oops']:
            with self.assertRaises(triage.InvalidReport):triage.parse_xml(xml([{'name':'a','duration':value}]))
    def test_duplicate_and_ambiguous_outcome(self):
        with self.assertRaises(triage.InvalidReport):triage.parse_xml(xml([{'name':'a'},{'name':'a'}]))
        with self.assertRaises(triage.InvalidReport):triage.parse_xml(b'<testsuite><testcase name="a"><failure/><skipped/></testcase></testsuite>')
    def test_duplicate_across_files(self):
        a=self.write([{'name':'a'}]);b=self.root/'other.xml';b.write_bytes(a.read_bytes())
        with self.assertRaises(triage.InvalidReport):triage.ingest(self.db,[a,b],build='b',sequence=1,job='j')
        self.assertFalse(self.db.exists())
    def test_resource_limits(self):
        for name,value,cases in [('MAX_BYTES',30,[{'name':'a'}]),('MAX_CASES',1,[{'name':'a'},{'name':'b'}]),('MAX_DEPTH',2,[{'name':'a'}])]:
            with patch.object(triage,name,value):
                with self.assertRaises(triage.InvalidReport):triage.parse_xml(xml(cases))
    def test_truncated_message_and_original_retained(self):
        self.add([{'name':'a','status':'failure','message':'x'*4090,'trace':'y'*40000}]);self.assertIn('Message truncated',self.rows()[0]['message']);self.assertGreater(len(triage.snapshot(self.db)['artifacts'][0]['content']),40000)
    def test_immutable_duplicate_and_conflict(self):
        self.assertTrue(self.add([{'name':'a'}])['ingested']);self.assertTrue(self.add([{'name':'a'}])['duplicate'])
        with self.assertRaises(triage.Conflict):self.add([{'name':'a','status':'failure'}])
        self.assertEqual(self.rows()[0]['status'],'pass')
    def test_replays_preserve_existing_database_bytes(self):
        self.add([{'name':'a'}]);before=self.db.read_bytes()
        self.assertTrue(self.add([{'name':'a'}])['duplicate'])
        self.assertEqual(before,self.db.read_bytes())
        with self.assertRaises(triage.Conflict):self.add([{'name':'a','status':'failure'}])
        self.assertEqual(before,self.db.read_bytes())
    def test_build_sequence_immutable(self):
        self.add([{'name':'a'}]);self.write([{'name':'b'}])
        for build,seq in [('b1',2),('other',1)]:
            with self.assertRaises(triage.Conflict):triage.ingest(self.db,[self.path],build=build,sequence=seq,job='j')
    def test_transaction_interruption_rolls_back(self):
        self.add([{'name':'baseline'}]);self.write([{'name':'a'},{'name':'b'}])
        with self.assertRaises(RuntimeError):triage.ingest(self.db,[self.path],build='b2',sequence=2,job='linux',fault_after=1)
        data=triage.snapshot(self.db);self.assertEqual(len(data['builds']),1);self.assertEqual(len(data['artifacts']),1)
    def test_dry_run_no_database(self):
        self.write([{'name':'a'}]);self.assertTrue(triage.ingest(self.db,[self.path],build='b',sequence=1,job='j',dry_run=True)['validated']);self.assertFalse(self.db.exists())
    def test_partial_requires_baseline(self):
        with self.assertRaises(triage.InvalidReport):self.add([{'name':'a'}],attempt=2,mode='partial')
    def test_partial_retry_preserves_membership(self):
        self.add([{'name':'a','status':'failure'},{'name':'b'}]);self.add([{'name':'a'}],attempt=2,mode='partial');rows={r['name']:r for r in self.rows()};self.assertEqual(rows['a']['category'],'retry pass');self.assertEqual(rows['b']['status'],'pass')
    def test_full_retry_omission_not_a_pass(self):
        self.add([{'name':'a'},{'name':'b','status':'failure'}]);self.add([{'name':'a'}],attempt=2);self.assertEqual({r['name']:r['category'] for r in self.rows()}['b'],'not reported')
    def test_new_repeated_recovered(self):
        self.add([{'name':'a'},{'name':'b','status':'failure'},{'name':'c','status':'error'}]);self.add([{'name':'a','status':'failure'},{'name':'b','status':'error'},{'name':'c'}],sequence=2);self.assertEqual({r['name']:r['category'] for r in self.rows('b2')},{'a':'new failure','b':'repeated failure','c':'recovered'})
    def test_missing_and_skipped_not_passes(self):
        self.add([{'name':'a'},{'name':'b'}]);self.add([{'name':'a','status':'skipped'}],sequence=2);self.assertEqual({r['name']:r['category'] for r in self.rows('b2')},{'a':'skipped','b':'not reported'})
    def test_future_identity_not_in_old_build(self):
        self.add([{'name':'a'}]);self.add([{'name':'a'},{'name':'future'}],sequence=2);self.assertEqual(len(self.rows('b1')),1)
    def test_identity_job_and_file(self):
        self.add([{'name':'a','file':'x.py'},{'name':'a','file':'y.py'}]);self.add([{'name':'a','file':'x.py'}],job='windows');self.assertEqual(len(self.rows()),3)
    def test_out_of_order_builds(self):
        self.add([{'name':'a','status':'failure'}],sequence=2);self.add([{'name':'a'}],sequence=1);self.assertEqual(self.rows('b2')[0]['category'],'new failure')
    def test_report_script_injection_escaped(self):
        payload='</script><script>alert("x")</script>';self.add([{'name':payload,'status':'failure','message':payload}]);text=render(triage.snapshot(self.db));self.assertNotIn(payload,text);self.assertIn('\\u003c/script',text);self.assertNotIn('innerHTML',text)
    def test_csv_formula_neutralized(self):
        self.add([{'name':'=2+2'}]);self.assertIn("'=2+2",triage.csv_text(triage.snapshot(self.db)['selections']['b1']))
    def test_export_deterministic_self_contained(self):
        self.add([{'name':'a'}]);triage.export(self.db,self.root/'report');first=(self.root/'report/index.html').read_bytes();triage.export(self.db,self.root/'report');self.assertEqual(first,(self.root/'report/index.html').read_bytes());self.assertNotIn(b'fetch(',first)
    def test_cli_exit_codes(self):
        self.add([{'name':'a','status':'failure'}]);out=io.StringIO()
        with contextlib.redirect_stdout(out),contextlib.redirect_stderr(out):
            self.assertEqual(triage.main(['check','--db',str(self.db)]),1);self.assertEqual(triage.main(['check','--db',str(self.db),'--build','unknown']),2)
        self.add([{'name':'a'}],sequence=2)
        with contextlib.redirect_stdout(out):self.assertEqual(triage.main(['check','--db',str(self.db)]),0)
    def test_lock_no_partial_write(self):
        self.add([{'name':'a'}]);connection=sqlite3.connect(self.db);connection.execute('BEGIN IMMEDIATE')
        try:
            with self.assertRaises(sqlite3.OperationalError):self.add([{'name':'b'}],sequence=2)
        finally:connection.rollback();connection.close()
        self.assertEqual(len(triage.snapshot(self.db)['builds']),1)
    def test_unrelated_database_not_modified(self):
        db=sqlite3.connect(self.db);db.execute('CREATE TABLE important(value TEXT)');db.execute("INSERT INTO important VALUES('retained')");db.commit();db.close()
        with self.assertRaises(triage.InvalidReport):self.add([{'name':'a'}])
        db=sqlite3.connect(self.db);self.assertEqual(db.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall(),[('important',)]);self.assertEqual(db.execute('SELECT * FROM important').fetchall(),[('retained',)]);db.close()
    def test_unsupported_version_byte_preservation(self):
        db=sqlite3.connect(self.db);db.execute('PRAGMA user_version=99');db.close();before=self.db.read_bytes()
        with self.assertRaises(triage.InvalidReport):self.add([{'name':'a'}])
        self.assertEqual(before,self.db.read_bytes())
    def test_concurrent_report_uses_one_read_snapshot(self):
        self.add([{'name':'a'}]);db=sqlite3.connect(self.db);db.execute('PRAGMA journal_mode=WAL');db.close()
        original=triage.connect;triggered=[];errors=[];owner=self
        class Observed(sqlite3.Connection):
            def execute(connection,sql,params=()):
                if sql.startswith('SELECT * FROM attempts') and not triggered:
                    triggered.append(True)
                    def write():
                        try:owner.add([{'name':'new'}],sequence=2)
                        except Exception as error:errors.append(str(error))
                    thread=threading.Thread(target=write);thread.start();thread.join(5)
                    owner.assertFalse(thread.is_alive());owner.assertFalse(errors)
                return super().execute(sql,params)
        def connect(path,create=True):
            if create:return original(path,create)
            connection=sqlite3.connect(Path(path).resolve().as_uri()+'?mode=ro',uri=True,isolation_level=None,factory=Observed);connection.row_factory=sqlite3.Row;return connection
        with patch.object(triage,'connect',connect):data=triage.snapshot(self.db)
        self.assertEqual(len(data['builds']),1);self.assertEqual(len(data['artifacts']),1);self.assertEqual(len(data['selections']['b1']['rows']),1)
        self.assertEqual(len(triage.snapshot(self.db)['builds']),2)
    def test_invalid_identity_symlink(self):
        self.write([{'name':'a'}]);link=self.root/'link.xml';link.symlink_to(self.path)
        with self.assertRaises(triage.InvalidReport):triage.read_artifacts([link])
        for seq in [0,-1,True]:
            with self.assertRaises(triage.InvalidReport):triage.ingest(self.db,[self.path],build='b',sequence=seq,job='j')
if __name__=='__main__':unittest.main(verbosity=2)