"""S3 protocol/configuration and real API/worker flows with a fake S3 transport.

No request is sent to a live provider and no real credential is required.
"""
import io
import importlib.util
from pathlib import Path
import time
import boto3
from botocore.exceptions import ClientError, EndpointConnectionError
from botocore.response import StreamingBody
from botocore.stub import Stubber
import pytest
from sqlalchemy import update
from app.config import settings
from app.storage import Storage, storage
from test_platform import client, KEY as API_KEY

NAMES=['input.SchDoc','input.PcbDoc','old.SchDoc','new.SchDoc','old.PcbDoc','new.PcbDoc',
       'comparison.json','pcb-comparison.json','attempt-1/data.json','attempt-1/drawing.svg',
       'attempt-1/old.svg','attempt-2/new.svg']


@pytest.fixture
def s3_config(tmp_path,monkeypatch):
    # Explicit fake credentials/client prevent the SDK's metadata discovery network calls.
    for name in ['STORAGE_BACKEND','S3_BUCKET','S3_PREFIX','S3_REGION','S3_ENDPOINT_URL','S3_ADDRESSING_STYLE',
                 'S3_CONNECT_TIMEOUT','S3_READ_TIMEOUT','S3_MAX_ATTEMPTS','S3_SERVER_SIDE_ENCRYPTION','S3_KMS_KEY_ID']:
        monkeypatch.delenv(name,raising=False)
    monkeypatch.setenv('STORAGE_DIR',str(tmp_path))
    monkeypatch.setenv('DATABASE_URL',f'sqlite:///{tmp_path}/storage.db')
    monkeypatch.setenv('APP_ENV','development')
    monkeypatch.setenv('STORAGE_BACKEND','s3');monkeypatch.setenv('S3_BUCKET','test-private')
    client=boto3.client('s3',region_name='us-east-1',aws_access_key_id='test',aws_secret_access_key='test')
    captures=[]
    def make_client(*args,**kwargs):captures.append((args,kwargs));return client
    monkeypatch.setattr(boto3,'client',make_client)
    settings.cache_clear();storage.cache_clear()
    yield client,captures,tmp_path
    settings.cache_clear();storage.cache_clear()


@pytest.mark.parametrize('name',NAMES)
@pytest.mark.parametrize('prefix',['','altium/production','/altium/production/'])
def test_s3_private_round_trip_contract(s3_config,monkeypatch,name,prefix):
    client,_,root=s3_config
    monkeypatch.setenv('S3_PREFIX',prefix);settings.cache_clear()
    key='a'*48+'/'+name;remote=prefix.strip('/')+'/'+key if prefix else key;payload=b'actual bytes'
    with Stubber(client) as stub:
        stub.add_response('put_object',{},dict(Bucket='test-private',Key=remote,Body=payload,ContentType='application/octet-stream'))
        stub.add_response('get_object',{'Body':StreamingBody(io.BytesIO(payload),len(payload))},dict(Bucket='test-private',Key=remote))
        stub.add_response('delete_object',{},dict(Bucket='test-private',Key=remote))
        service=Storage();service.put(key,payload);assert service.get(key)==payload;service.delete(key)
        stub.assert_no_pending_responses()
    assert not (root/'objects').exists()


@pytest.mark.parametrize('mode,kms,options',[
    ('','',{}),('AES256','',{'ServerSideEncryption':'AES256'}),
    ('aws:kms','alias/altium',{'ServerSideEncryption':'aws:kms','SSEKMSKeyId':'alias/altium'}),
])
def test_encryption_config_and_sdk_transport(s3_config,monkeypatch,mode,kms,options):
    client,captures,_=s3_config
    for key,value in {'S3_ENDPOINT_URL':'https://objects.example.test','S3_REGION':'test-region',
                      'S3_ADDRESSING_STYLE':'path','S3_CONNECT_TIMEOUT':'7','S3_READ_TIMEOUT':'31','S3_MAX_ATTEMPTS':'4',
                      'S3_SERVER_SIDE_ENCRYPTION':mode,'S3_KMS_KEY_ID':kms}.items():monkeypatch.setenv(key,value)
    settings.cache_clear();key='a'*48+'/input.PcbDoc'
    with Stubber(client) as stub:
        stub.add_response('put_object',{},dict(Bucket='test-private',Key=key,Body=b'pcb',ContentType='application/octet-stream',**options))
        Storage().put(key,b'pcb');stub.assert_no_pending_responses()
    args,kwargs=captures[0];assert args==('s3',)
    assert kwargs['endpoint_url']=='https://objects.example.test' and kwargs['region_name']=='test-region'
    config=kwargs['config'];assert config.signature_version=='s3v4' and config.s3['addressing_style']=='path'
    assert config.connect_timeout==7 and config.read_timeout==31 and config.retries['total_max_attempts']==4
    assert not {'aws_access_key_id','aws_secret_access_key'} & kwargs.keys()


@pytest.mark.parametrize('key,value',[
    ('STORAGE_BACKEND','unknown'),('S3_BUCKET',''),('S3_PREFIX','../escape'),('S3_PREFIX','nested/../escape'),
    ('S3_PREFIX','a//b'),('S3_PREFIX','bad\\prefix'),('S3_PREFIX','bad\nname'),('S3_PREFIX','x'*513),
    ('S3_ADDRESSING_STYLE','invalid'),('S3_CONNECT_TIMEOUT','0'),('S3_READ_TIMEOUT','-1'),('S3_MAX_ATTEMPTS','0'),
    ('S3_SERVER_SIDE_ENCRYPTION','public'),('S3_KMS_KEY_ID','alias/missing-encryption-mode'),
    ('S3_ENDPOINT_URL','file:///tmp/data'),('S3_ENDPOINT_URL','https://user:secret@example.test'),
    ('S3_ENDPOINT_URL','https://example.test/?token=secret'),
])
def test_invalid_s3_configuration_fails_closed(s3_config,monkeypatch,key,value):
    monkeypatch.setenv(key,value);settings.cache_clear()
    with pytest.raises(RuntimeError):settings()


def test_auto_backend_compatibility_and_no_local_directory_with_postgres(s3_config,monkeypatch):
    _,_,root=s3_config
    monkeypatch.delenv('STORAGE_BACKEND');settings.cache_clear()
    assert settings().storage_backend=='s3'
    monkeypatch.setenv('S3_BUCKET','');settings.cache_clear()
    assert settings().storage_backend=='local'
    monkeypatch.setenv('STORAGE_BACKEND','s3');monkeypatch.setenv('S3_BUCKET','test-private')
    monkeypatch.setenv('DATABASE_URL','postgresql+psycopg://unused:unused@localhost/unused')
    monkeypatch.setenv('STORAGE_DIR',str(root/'unused-local-dir'));settings.cache_clear()
    assert settings().storage_backend=='s3' and not (root/'unused-local-dir').exists()


def test_s3_errors_never_fall_back_to_disk(s3_config,monkeypatch):
    client,_,root=s3_config
    key='a'*48+'/input.SchDoc'
    with Stubber(client) as stub:
        stub.add_client_error('put_object',service_error_code='AccessDenied',http_status_code=403)
        stub.add_client_error('get_object',service_error_code='AccessDenied',http_status_code=403)
        stub.add_client_error('delete_object',service_error_code='AccessDenied',http_status_code=403)
        service=Storage()
        for operation in (lambda:service.put(key,b'x'),lambda:service.get(key),lambda:service.delete(key)):
            with pytest.raises(ClientError):operation()
    assert not (root/'objects').exists()
    for key in ['../input.SchDoc','a'*48+'/input.SchDoc/other','foreign/'+'a'*48+'/input.PcbDoc']:
        with pytest.raises(ValueError):service.object_key(key)


class MemoryS3:
    """Transport fake only: all file validation/parsing/rendering/worker logic is real."""
    def __init__(self):self.objects={};self.content_types={}
    def put_object(self,Bucket,Key,Body,ContentType,**kwargs):
        self.objects[(Bucket,Key)]=bytes(Body);self.content_types[(Bucket,Key)]=ContentType
        assert 'ACL' not in kwargs
    def get_object(self,Bucket,Key):
        if (Bucket,Key) not in self.objects:
            raise ClientError({'Error':{'Code':'NoSuchKey'},'ResponseMetadata':{'HTTPStatusCode':404}},'GetObject')
        data=self.objects[(Bucket,Key)];return {'Body':StreamingBody(io.BytesIO(data),len(data))}
    def delete_object(self,Bucket,Key):self.objects.pop((Bucket,Key),None)


@pytest.mark.parametrize('kind',['schematic_render','schematic_comparison','pcb_render','pcb_comparison'])
def test_api_worker_read_and_expiry_cleanup_use_only_prefixed_s3(client,monkeypatch,ascii_schematic,kind):
    from app.db import Job,claim,sessions
    from app.worker import process
    from app.cleanup import cleanup
    from pcb_fixture import pcb
    fake=MemoryS3();monkeypatch.setattr(boto3,'client',lambda *a,**kw:fake)
    monkeypatch.setenv('STORAGE_BACKEND','s3');monkeypatch.setenv('S3_BUCKET','test-private');monkeypatch.setenv('S3_PREFIX','altium/test')
    settings.cache_clear();storage.cache_clear()
    pcb_kind=kind.startswith('pcb');comparison=kind.endswith('comparison')
    suffix='PcbDoc' if pcb_kind else 'SchDoc';raw=pcb() if pcb_kind else ascii_schematic
    route=('/api/v1/pcbs/comparisons' if comparison else '/api/v1/pcbs/render') if pcb_kind else ('/api/v1/comparisons' if comparison else '/api/v1/schematics/render')
    files={f'{side}_file':(side+'.'+suffix,raw) for side in ('old','new')} if comparison else {'file':('input.'+suffix,raw)}
    response=client.post(route,headers={'X-API-Key':API_KEY},files=files)
    assert response.status_code==202,response.text
    created=response.json();identifier=created['id'];headers={'X-Viewer-Token':created['viewer_token']}
    fake.objects[('test-private','other-prefix/keep')]=b'other service data'
    process(claim());status=client.get(created['status_url'],headers=headers).json()
    assert status['status']=='completed',status
    base=created['status_url'].removesuffix('/status')
    assert client.get(base+('/svg/old' if comparison else '/svg'),headers=headers).status_code==200
    assert client.get(created['status_url']).status_code==404
    if comparison:assert status['total_changed']==0
    else:assert client.get(base+'/data',headers=headers).status_code==200
    assert all(key.startswith('altium/test/'+identifier+'/') for bucket,key in fake.objects if key!='other-prefix/keep')
    assert not (settings().storage_dir/'objects').exists()
    with sessions().begin() as db:db.execute(update(Job).where(Job.id==identifier).values(expires_at=time.time()-1))
    assert cleanup()==1
    assert fake.objects=={('test-private','other-prefix/keep'):b'other service data'}


def test_migration_copies_verifies_skips_identical_and_refuses_overwrite(s3_config,monkeypatch,tmp_path):
    fake=MemoryS3();monkeypatch.setattr(boto3,'client',lambda *a,**kw:fake)
    monkeypatch.setenv('S3_PREFIX','altium/test');settings.cache_clear()
    script=Path(__file__).resolve().parents[2]/'scripts/migrate-to-s3.py'
    spec=importlib.util.spec_from_file_location('migration',script);module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
    key='a'*48+'/input.PcbDoc';source=tmp_path/'source';file=source/'objects'/key
    file.parent.mkdir(parents=True);file.write_bytes(b'original')
    assert module.migrate(source)=={'copied_and_verified':1,'already_identical':0}
    assert file.read_bytes()==b'original'
    assert module.migrate(source)=={'copied_and_verified':0,'already_identical':1}
    fake.objects[('test-private','altium/test/'+key)]=b'different'
    with pytest.raises(ValueError,match='refusing to overwrite'):module.migrate(source)
    assert file.read_bytes()==b'original'