import modulus.conan
import modulus.build_context
import modulus.container
import modulus.packaging
import modulus.process
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.build_context.build_context(root, modulus.config.normalize({'package': {'name': 'sample'}}))
    with patch.dict(os.environ, credentials):
      for env in (modulus.bootstrap.fetch_env(root), modulus.process.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.build_context.build_context(Path('/tmp/project'), modulus.config.normalize({'package': {'name': 'sample'}}))
    with patch.object(modulus.conan, 'configure_conan_remote', return_value=0), patch.object(modulus.packaging, 'upload_all_packages', side_effect=RuntimeError('upload failed')), patch.object(modulus.process, 'run', return_value=0) as run:
      with self.assertRaisesRegex(RuntimeError, 'upload failed'):
        modulus.packaging.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.build_context.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.container.recipe_digest(context)
      context.docker_inputs[-1].write_text('new')
      self.assertNotEqual(before, modulus.container.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.container.docker_socket_args(context, command), [])
        self.assertIn('/var/run/docker.sock:/var/run/docker.sock', modulus.container.docker_socket_args(context, 'package'))