prj-forgejo-org-refactor/tests/test_inventory.py

import importlib.util
import io
import json
import os
from pathlib import Path
import subprocess
import tempfile
import unittest
from unittest.mock import Mock, patch
import urllib.error
import urllib.request
import zipfile


def module(name):
    spec=importlib.util.spec_from_file_location(name,Path(__file__).parents[1]/'tools'/f'{name}.py')
    m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m);return m

inv=module('inventory');verify=module('verify_inventory')

class InventoryTests(unittest.TestCase):
    def test_scan_emits_reference_only_and_skips_sensitive_files(self):
        with tempfile.TemporaryDirectory() as d:
            p=Path(d)
            subprocess.run(['git','init','-q',d],check=True)
            (p/'values.yaml').write_text('image: https://user:do-not-emit@forgejo.example/coulomb/example:tag?token=also-private\n')
            (p/'.env').write_text('SECRET=coulomb/private-secret\n')
            (p/'link').symlink_to(p/'values.yaml')
            subprocess.run(['git','-C',d,'add','.'],check=True)
            subprocess.run(['git','-C',d,'-c','user.name=Test','-c','user.email=test@example.invalid','commit','-qm','fixture'],check=True)
            result=inv.scan(p)
            self.assertEqual([r['reference'] for r in result['references']],['coulomb/example'])
            self.assertNotIn('do-not-emit',str(result));self.assertNotIn('also-private',str(result))
            self.assertNotIn('private-secret',str(result))
            self.assertEqual({e['reason'] for e in result['exclusions']},{'sensitive-or-encrypted','symlink-not-followed'})
            self.assertEqual(result,inv.scan(p))

    def test_paginated_server_read_does_not_stop_at_first_page(self):
        original=inv.get
        try:
            inv.get=lambda url: [{'id':1}] if 'page=1' in url else [{'id':2}] if 'page=2' in url else []
            self.assertEqual(inv.pages('/repos'),[{'id':1},{'id':2}])
            inv.get=lambda url:[{'id':1}]
            with self.assertRaises(RuntimeError):inv.pages('/repos')
        finally:inv.get=original

    def test_empty_partial_inventory_cannot_pass(self):
        result=verify.check({'g1_complete':False,'local_repositories':[]},{'approved':False,'assignments':[]})
        self.assertFalse(result['clean'])

    def test_retained_org_and_stale_moved_path(self):
        i={'g1_complete':True,'local_repositories':[{'name':'consumer','references':[{'reference':'coulomb/example','categories':['historical-or-planning']}]}]}
        m={'approved':True,'assignments':[{'repository':'example','proposed_org':'coulomb'}]}
        self.assertTrue(verify.check(i,m)['clean'])
        m['assignments'][0]['proposed_org']='railiance'
        self.assertFalse(verify.check(i,m)['clean'])

    def test_authenticated_reads_use_get_and_are_scoped_to_forge(self):
        response = Mock()
        response.read.return_value = b'{"id": 42}'
        opener = Mock()
        opener.open.return_value.__enter__ = Mock(return_value=response)
        opener.open.return_value.__exit__ = Mock(return_value=False)
        with patch.object(inv.urllib.request, 'build_opener', return_value=opener):
            self.assertEqual(inv.get(inv.BASE + '/api/v1/repos/coulomb/private', token='fixture-only'), {'id':42})
            request = opener.open.call_args.args[0]
            self.assertEqual(request.get_method(), 'GET')
            self.assertEqual(request.get_header('Authorization'), 'token fixture-only')
            for url in ['http://127.0.0.1:8000/repos/', 'https://other.invalid/repos/',
                        'http://forgejo.coulomb.social/repos/']:
                with self.assertRaises(RuntimeError):
                    inv.get(url, token='fixture-only')
            inv.get('http://127.0.0.1:8000/repos/')
            self.assertIsNone(opener.open.call_args.args[0].get_header('Authorization'))

    def test_authenticated_redirects_cannot_leak_token_or_downgrade(self):
        req = urllib.request.Request(inv.BASE + '/start', headers={'Authorization':'token fixture-only'})
        handler = inv.ScopedRedirects(True)
        redirect = handler.redirect_request(req, None, 302, 'Found', {}, inv.BASE + '/next')
        self.assertEqual(redirect.get_header('Authorization'), 'token fixture-only')
        for url in ['https://other.invalid/next', 'http://forgejo.coulomb.social/next',
                    'https://user@forgejo.coulomb.social/next']:
            with self.assertRaisesRegex(RuntimeError, '^Authenticated forge redirect refused$'):
                handler.redirect_request(req, None, 302, 'Found', {}, url)

    def test_transport_errors_do_not_emit_credential_or_raw_response(self):
        for error, message in [
            (urllib.error.URLError('fixture-only raw error'), 'Inventory read failed'),
            (urllib.error.HTTPError(inv.BASE, 403, 'fixture-only raw response', {}, None), 'HTTP 403')]:
            opener = Mock()
            opener.open.side_effect = error
            with patch.object(inv.urllib.request, 'build_opener', return_value=opener):
                with self.assertRaises(RuntimeError) as caught:
                    inv.get(inv.BASE + '/api/v1/repos', token='fixture-only')
                self.assertEqual(str(caught.exception), message)

    def test_token_is_forwarded_to_all_metadata_reads_but_not_hub(self):
        calls = []
        def fake_get(url, *, token=None):
            calls.append((url,token))
            if url.endswith('/repos/'):
                return []
            if '/orgs/' in url and 'page=1' in url:
                return [{'id':42, 'name':inv.SELF.name, 'full_name':'coulomb/'+inv.SELF.name, 'private':True}]
            if '/orgs/' in url or '/packages/' in url:
                return []
            return {'version':'fixture'}
        with tempfile.TemporaryDirectory() as root, patch.object(inv,'get',side_effect=fake_get):
            result=inv.collect(Path(root),token='fixture-only')
        self.assertFalse(result['g1_complete'])
        self.assertEqual(result['forge']['authentication'],'authenticated-read')
        self.assertTrue(result['server_repositories'][0]['private'])
        for url,token in calls:
            self.assertEqual(token, 'fixture-only' if url.startswith(inv.BASE) else None)
        self.assertNotIn('fixture-only',json.dumps(result))

    def test_missing_requested_credential_refuses_before_collection(self):
        with tempfile.TemporaryDirectory() as root:
            output=Path(root)/'snapshot.json'
            env=dict(os.environ)
            env.pop('ORGREF_TEST_MISSING_TOKEN',None)
            result=subprocess.run([
                'python3',str(Path(__file__).parents[1]/'tools/inventory.py'),
                '--output',str(output),'--forge-token-env','ORGREF_TEST_MISSING_TOKEN'],
                env=env,capture_output=True,text=True)
            self.assertEqual(result.returncode,2)
            self.assertFalse(output.exists())

    def test_private_archive_uses_the_same_scoped_read_credential(self):
        raw=io.BytesIO()
        with zipfile.ZipFile(raw,'w') as archive:
            archive.writestr('private/values.yaml','image: forgejo.example/coulomb/example:tag\n')
        with patch.object(inv,'get',return_value={'commit':{'id':'abc'}}) as metadata, \
             patch.object(inv,'read_bytes',return_value=raw.getvalue()) as archive_read:
            result=inv.scan_archive({'name':'private','default_branch':'main'},token='fixture-only')
        self.assertEqual(metadata.call_args.kwargs['token'],'fixture-only')
        self.assertEqual(archive_read.call_args.kwargs,{'token':'fixture-only','limit':50_000_001})
        self.assertEqual([r['reference'] for r in result['references']],['coulomb/example'])
        self.assertNotIn('fixture-only',json.dumps(result))

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