import ast
import json
import os
from pathlib import Path
import subprocess
import tempfile
import unittest
from unittest.mock import patch

import modulus.bootstrap
import modulus.cli
import modulus.config
import modulus.export

class BootstrapTests(unittest.TestCase):
  def test_recipe_exports_prune_generated_trees(self):
    workspace = Path(modulus.bootstrap.__file__).parents[2]
    for name in ('modulus', 'ligo', 'noto', 'luma', 'sonus', 'product-vamp'):
      recipe = workspace / name / 'conanfile.py'
      if not recipe.exists():
        continue
      fields = {target.id: ast.literal_eval(node.value) for node in ast.walk(ast.parse(recipe.read_text())) if isinstance(node, ast.Assign) for target in node.targets if isinstance(target, ast.Name) and target.id in {'exports', 'exports_sources'}}
      for field in ('exports', 'exports_sources'):
        self.assertTrue({'!build', '!.git', '!test_package/build'}.issubset(fields[field]), f'{name}: {field}')
  def test_consumers_use_canonical_bootstrap(self):
    canonical = Path(modulus.bootstrap.__file__)
    for name in ('ligo', 'noto', 'luma', 'sonus', 'product-vamp'):
      consumer = canonical.parents[2] / name / 'modulus.py'
      if consumer.exists():
        self.assertEqual(canonical.read_bytes(), consumer.read_bytes(), name)
  def test_read_environments_strip_all_credential_forms(self):
    credentials = dict.fromkeys(('CONAN_USER', 'CONAN_TOKEN', 'CONAN_LOGIN_USERNAME', 'CONAN_PASSWORD', 'CONAN_LOGIN_USERNAME_KHODO', 'CONAN_PASSWORD_KHODO'), 'secret')
    root = Path('/tmp/project')
    context = modulus.cli.build_context(root, modulus.config.normalize({'package': {'name': 'sample'}}))
    with patch.dict(os.environ, credentials):
      for env in (modulus.bootstrap.fetch_env(root), modulus.cli.subprocess_environment(context, ['conan', 'install'])):
        self.assertFalse(set(credentials) & env.keys())
        self.assertEqual(env['CONAN_NON_INTERACTIVE'], '1')
        self.assertEqual(env['CONAN_HOME'], str(root / 'build/build_common/conan_cache'))
  def test_update_refreshes_runtime_and_offline_stays_offline(self):
    root = Path('/tmp/project')
    with patch.object(modulus.bootstrap, 'runtime_ready', return_value=True), patch.object(modulus.bootstrap, 'materialize_runtime') as refresh:
      modulus.bootstrap.ensure_runtime(root)
      refresh.assert_not_called()
      modulus.bootstrap.ensure_runtime(root, update=True, no_remote=True)
      refresh.assert_called_once_with(root, no_remote=True)
    with patch.object(modulus.bootstrap, 'run_fetch', return_value=subprocess.CompletedProcess([], 0, '{}')) as fetch:
      with self.assertRaisesRegex(SystemExit, 'local Conan cache'):
        modulus.bootstrap.find_modulus(root)
      self.assertNotIn('-r', fetch.call_args.args[1])
  def test_fetch_failure_preserves_runtime(self):
    with tempfile.TemporaryDirectory() as directory:
      root = Path(directory)
      runtime = root / modulus.bootstrap.RUNTIME_DIR / 'modulus/cli.py'
      runtime.parent.mkdir(parents=True)
      runtime.write_text('old runtime')
      with patch.object(modulus.bootstrap, 'remote_config', return_value=('khodo', 'https://example.test')), patch.object(modulus.bootstrap, 'configure_remote'), patch.object(modulus.bootstrap, 'find_modulus', side_effect=SystemExit(1)):
        with self.assertRaises(SystemExit):
          modulus.bootstrap.materialize_runtime(root)
      self.assertEqual(runtime.read_text(), 'old runtime')
  def test_offline_bootstrap_never_builds_a_fetch_image(self):
    with patch.object(modulus.bootstrap.subprocess, 'run', return_value=subprocess.CompletedProcess([], 1)), patch.object(modulus.bootstrap, 'run') as build:
      with self.assertRaisesRegex(SystemExit, 'offline bootstrap requires'):
        modulus.bootstrap.ensure_fetch_image(offline=True)
      build.assert_not_called()
  def test_export_rolls_back_when_editor_refresh_fails(self):
    with tempfile.TemporaryDirectory() as directory:
      root = Path(directory)
      (root / 'modulus.json').write_text(json.dumps({'package': {'name': 'sample'}}))
      runtime = root / modulus.export.runtime_dir / 'modulus/cli.py'
      runtime.parent.mkdir(parents=True)
      runtime.write_text('old runtime')
      package = root / 'package'
      (package / 'modulus').mkdir(parents=True)
      (package / 'modulus/cli.py').write_text('new runtime')
      (package / 'tools').mkdir()
      with patch.object(modulus.export, 'sync_editor_files', side_effect=OSError('editor is read-only')):
        with self.assertRaisesRegex(OSError, 'read-only'):
          modulus.export.sync_project(root, package)
      self.assertEqual(runtime.read_text(), 'old runtime')
      self.assertFalse((root / modulus.export.tools_dir).exists())
  def test_public_remote_error_is_noninteractive_and_does_not_delete_packages(self):
    completed = subprocess.CompletedProcess([], 1, '', '401 Unauthorized')
    with patch.object(modulus.bootstrap, 'direct_mode', return_value=True), patch.object(modulus.bootstrap.subprocess, 'run', return_value=completed) as run:
      with self.assertRaises(SystemExit):
        modulus.bootstrap.find_modulus(Path('/tmp/project'), 'khodo')
      self.assertEqual(run.call_args.args[0][:2], ['conan', 'list'])
      self.assertEqual(run.call_args.kwargs['env']['CONAN_NON_INTERACTIVE'], '1')
  def test_export_stages_all_payloads_before_replacing_runtime(self):
    with tempfile.TemporaryDirectory() as directory:
      root = Path(directory)
      (root / 'modulus.json').write_text(json.dumps({'package': {'name': 'sample'}}))
      runtime = root / modulus.export.runtime_dir / 'modulus/cli.py'
      runtime.parent.mkdir(parents=True)
      runtime.write_text('old runtime')
      package = root / 'package'
      (package / 'modulus').mkdir(parents=True)
      (package / 'modulus/cli.py').write_text('new runtime')
      with self.assertRaisesRegex(RuntimeError, 'missing tools'):
        modulus.export.sync_project(root, package)
      self.assertEqual(runtime.read_text(), 'old runtime')
  def test_upload_exception_still_logs_out(self):
    context = modulus.cli.build_context(Path('/tmp/project'), modulus.config.normalize({'package': {'name': 'sample'}}))
    with patch.object(modulus.cli, 'configure_conan_remote', return_value=0), patch.object(modulus.cli, 'upload_all_packages', side_effect=RuntimeError('upload failed')), patch.object(modulus.cli, 'run', return_value=0) as run:
      with self.assertRaisesRegex(RuntimeError, 'upload failed'):
        modulus.cli.upload_packages(context)
      run.assert_called_once_with(context, ['conan', 'remote', 'logout', 'khodo'])
  def test_socket_is_scoped_and_docker_inputs_change_fingerprint(self):
    with tempfile.TemporaryDirectory() as directory:
      root = Path(directory)
      config = modulus.config.normalize({'package': {'name': 'sample'}, 'tooling': {'docker': {'dockerfile': 'package/build/Dockerfile', 'inputs': ['requirements.txt'], 'socket_commands': ['build', 'package']}}})
      context = modulus.cli.build_context(root, config)
      context.dockerfile.parent.mkdir(parents=True)
      context.dockerfile.write_text('FROM scratch')
      context.docker_inputs[0].parent.mkdir(parents=True)
      for path in context.docker_inputs:
        path.write_text('old')
      before = modulus.clangd.recipe_digest(context)
      context.docker_inputs[-1].write_text('new')
      self.assertNotEqual(before, modulus.clangd.recipe_digest(context))
      with patch.object(Path, 'is_socket', return_value=True), patch.object(Path, 'stat') as stat:
        stat.return_value.st_gid = 123
        for command in ('configure', 'test', 'lsp', 'clangd', 'format'):
          self.assertEqual(modulus.cli.docker_socket_args(context, command), [])
        self.assertIn('/var/run/docker.sock:/var/run/docker.sock', modulus.cli.docker_socket_args(context, 'package'))