import concurrent.futures
import copy
from pathlib import Path
import sqlite3
import tempfile
import unittest

from profiles import asset_evidence as a
from profiles.asset_catalog import AssetCatalog
from test_asset_evidence import asset_fixture,manifest,bundle,ref,snapshot,NOW,ORIGIN,later


class AssetCatalogTests(unittest.TestCase):
    def setUp(self):
        self.temp=tempfile.TemporaryDirectory();self.path=Path(self.temp.name)/'catalog.sqlite3'
        self.meta,self.pol,self.op=asset_fixture();self.m=manifest(self.meta,self.pol,self.op)
        self.raw=bundle([self.m],self.meta,self.pol);self.snap=snapshot(self.meta,self.pol)
        self.trust=a.AssetTrust({ORIGIN:self.snap},{})
        self.clock=lambda:NOW;self.catalog=AssetCatalog(self.path)
    def tearDown(self):self.temp.cleanup()

    def test_restart_preserves_exact_original_manifest_and_closed_dependencies(self):
        self.catalog.retain(self.raw,trust=self.trust,now=self.clock)
        child=manifest(self.meta,self.pol,self.op,number=2,parents=[ref(self.m)])
        whole=bundle([self.m,child],self.meta,self.pol)
        self.catalog.retain(whole,trust=self.trust,now=self.clock)
        restarted=AssetCatalog(self.path);saved=restarted.get_manifest(self.m['version_id'])
        self.assertEqual(saved['payload_bytes'],a.raw(self.m));self.assertFalse(saved['resource_permission_granted'])
        reconstructed=restarted.bundle_for([ref(child)])
        self.assertEqual(len(a.verify_closure(reconstructed,trust=self.trust,now=self.clock).manifests),2)

    def test_same_version_cannot_change_after_restart(self):
        self.catalog.retain(self.raw,trust=self.trust,now=self.clock)
        changed=copy.deepcopy(self.m);changed['representations'][0]['languages']=[]
        with self.assertRaisesRegex(a.AssetError,'version redefined'):
            AssetCatalog(self.path).retain(bundle([changed],self.meta,self.pol),trust=self.trust,now=self.clock)
        self.assertEqual(self.catalog.get_manifest(self.m['version_id'])['payload_bytes'],a.raw(self.m))

    def test_concurrent_exact_retries_keep_one_bundle_and_original_report(self):
        def retain(_):return self.catalog.retain(self.raw,trust=self.trust,now=self.clock)
        with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:results=list(pool.map(retain,range(3)))
        self.assertEqual(results,[results[0]]*3)
        with sqlite3.connect(self.path) as db:
            self.assertEqual(db.execute('SELECT count(*) FROM catalog_bundles').fetchone()[0],1)
            self.assertEqual(db.execute('SELECT count(*) FROM catalog_manifests').fetchone()[0],1)

    def test_late_current_authority_failure_rolls_back_all_material(self):
        ticks=iter([NOW,NOW,later(6),later(6)])
        with self.assertRaises(a.AssetError):self.catalog.retain(self.raw,trust=self.trust,now=lambda:next(ticks))
        with sqlite3.connect(self.path) as db:
            for table in ('catalog_bundles','catalog_manifests','catalog_documents'):
                self.assertEqual(db.execute('SELECT count(*) FROM '+table).fetchone()[0],0)

    def test_exact_retry_rechecks_current_compromise(self):
        self.catalog.retain(self.raw,trust=self.trust,now=self.clock)
        revoked=copy.deepcopy(self.meta);revoked['revision']=2
        revoked['services'][0]['signing_keys'][0].update(state='revoked',revoked_at=NOW)
        trust=a.AssetTrust({ORIGIN:snapshot(revoked,self.pol)},{a.digest(self.snap.authority_bytes):self.snap.authority_bytes})
        with self.assertRaises(a.AssetError):self.catalog.retain(self.raw,trust=trust,now=self.clock)

    def test_private_paths_and_operational_table_separation(self):
        other=Path(self.temp.name)/'service.sqlite3'
        with sqlite3.connect(other) as db:db.execute('CREATE TABLE agreements(id TEXT)')
        other.chmod(0o600)
        with self.assertRaises(a.AssetError):AssetCatalog(other)
        link=Path(self.temp.name)/'linked.sqlite3';link.symlink_to(self.path)
        with self.assertRaises(OSError):AssetCatalog(link)

    def test_catalog_entries_are_immutable_and_qualified_refs_are_exact(self):
        self.catalog.retain(self.raw,trust=self.trust,now=self.clock)
        with sqlite3.connect(self.path) as db:
            with self.assertRaises(sqlite3.IntegrityError):db.execute('DELETE FROM catalog_manifests')
        changed=ref(self.m);changed['representation_id']='00000000-0000-4000-8000-000000999999'
        with self.assertRaises(a.AssetError):self.catalog.bundle_for([changed])


if __name__=='__main__':unittest.main()