#!/usr/bin/env python3
"""Repository-owned sdkharness customer-health lifecycle for b2sdk."""

from __future__ import annotations

import io
import os
import sys
import uuid
from pathlib import Path
from typing import Callable, Mapping
from urllib.parse import urlsplit

LEVEL = 'health'
SCENARIO = 'golden-path'
SIMULATOR_CREDENTIAL = ('test-key-id', 'test-key')
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]

sys.path.insert(0, str(Path(__file__).resolve().parent / 'lib'))
from loopback_guard import scrub_proxy_environment  # noqa: E402


class CheckFailure(Exception):
    """A failed lifecycle step with a credential-safe reason."""

    def __init__(self, step: str, cause: BaseException | str) -> None:
        super().__init__(step)
        self.step = step
        self.detail = cause if isinstance(cause, str) else type(cause).__name__


def checked(step: str, action: Callable):
    try:
        return action()
    except CheckFailure:
        raise
    except Exception as error:
        raise CheckFailure(step, error) from error


def validate_environment(environment: Mapping[str, str]) -> tuple[str, str, str, str]:
    if environment.get('SDKHARNESS_TEST_LEVEL', LEVEL) != LEVEL:
        raise CheckFailure('configuration', 'unexpected test level')
    if environment.get('SDKHARNESS_SCENARIO', SCENARIO) != SCENARIO:
        raise CheckFailure('configuration', 'unexpected scenario')

    simulator_url = environment.get('SDKHARNESS_SIMULATOR_URL', '')
    if not simulator_url or environment.get('HEALTHCHECK_REALM_URL') != simulator_url:
        raise CheckFailure('configuration', 'simulator URL mismatch')

    try:
        parsed = urlsplit(simulator_url)
        port = parsed.port
    except ValueError as error:
        raise CheckFailure('configuration', 'invalid simulator URL') from error
    if (
        parsed.scheme != 'http'
        or parsed.hostname != '127.0.0.1'
        or port is None
        or parsed.username is not None
        or parsed.password is not None
        or parsed.path not in {'', '/'}
        or parsed.query
        or parsed.fragment
    ):
        raise CheckFailure('configuration', 'simulator URL must be loopback HTTP')

    key_id = environment.get('B2_TEST_APPLICATION_KEY_ID', '')
    application_key = environment.get('B2_TEST_APPLICATION_KEY', '')
    bucket_name = environment.get('B2_BUCKET_NAME', '')
    if not key_id or not application_key or not bucket_name:
        raise CheckFailure('configuration', 'required simulator input is missing')
    # Only the simulator's fixed test credential is ever used. A real key in the
    # environment is refused before it can reach the SDK (and so a loopback
    # listener); the supplied values are never echoed.
    if (key_id, application_key) != SIMULATOR_CREDENTIAL:
        raise CheckFailure('configuration', 'only the fixed simulator credential is accepted')
    return simulator_url, SIMULATOR_CREDENTIAL[0], SIMULATOR_CREDENTIAL[1], bucket_name


def repository_api_factory():
    sys.path.insert(0, str(REPOSITORY_ROOT))
    import b2sdk
    from b2sdk.v3 import B2Api, InMemoryAccountInfo

    module_path = Path(b2sdk.__file__).resolve()
    try:
        module_path.relative_to(REPOSITORY_ROOT)
    except ValueError as error:
        raise CheckFailure('setup', 'b2sdk was not imported from this checkout') from error
    return B2Api(InMemoryAccountInfo())


def run_health(
    environment: Mapping[str, str],
    *,
    api_factory: Callable = repository_api_factory,
    object_name: str | None = None,
) -> None:
    simulator_url, key_id, application_key, bucket_name = validate_environment(environment)
    api = checked('load SDK', api_factory)
    checked(
        'authenticate',
        lambda: api.authorize_account(key_id, application_key, realm=simulator_url),
    )
    bucket = checked('bucket', lambda: api.get_bucket_by_name(bucket_name))

    name = object_name or f'sdkharness-health-check/{uuid.uuid4().hex}.txt'
    payload = b'b2-sdk-python health check ' + name.encode('ascii')
    uploaded = None
    failure = None
    try:
        uploaded = checked('upload', lambda: bucket.upload_bytes(payload, name))
        downloaded = io.BytesIO()
        checked('download', lambda: bucket.download_file_by_name(name).save(downloaded))
        if downloaded.getvalue() != payload:
            raise CheckFailure('verify bytes', 'download differed from upload')

        if not checked(
            'list',
            lambda: any(
                version.file_name == name for version, _ in bucket.ls(name, recursive=True)
            ),
        ):
            raise CheckFailure('list', 'uploaded object was not listed')

        checked('delete', lambda: bucket.delete_file_version(uploaded.id_, name))
        uploaded = None
        if checked(
            'confirm gone',
            lambda: any(
                version.file_name == name for version, _ in bucket.ls(name, recursive=True)
            ),
        ):
            raise CheckFailure('confirm gone', 'deleted object was still listed')
    except CheckFailure as error:
        failure = error
    finally:
        if uploaded is not None:
            try:
                bucket.delete_file_version(uploaded.id_, name)
            except Exception as error:
                if failure is None:
                    failure = CheckFailure('cleanup', error)
    if failure is not None:
        raise failure


def main() -> int:
    scrub_proxy_environment(os.environ)
    try:
        run_health(os.environ)
    except CheckFailure as error:
        print(f'SDKHARNESS_RESULT\t{LEVEL}\t{SCENARIO}\tFAIL\t{error.step}: {error.detail}')
        return 1
    except Exception as error:
        print(f'SDKHARNESS_RESULT\t{LEVEL}\t{SCENARIO}\tFAIL\tsetup: {type(error).__name__}')
        return 1
    print(f'SDKHARNESS_RESULT\t{LEVEL}\t{SCENARIO}\tPASS\t-')
    return 0


if __name__ == '__main__':
    sys.exit(main())
