SteamFramelyHomebrew--framely/tests/test_diagnostics.py

import base64
import importlib.util
import io
import json
import pathlib
import tempfile
import unittest
from unittest import mock
from types import SimpleNamespace
import subprocess
import zipfile

spec = importlib.util.spec_from_file_location('diagnostics', pathlib.Path(__file__).parents[1] / 'tools/export-diagnostics.py')
diag = importlib.util.module_from_spec(spec)
spec.loader.exec_module(diag)

class DiagnosticsTests(unittest.TestCase):
    def archive(self, root):
        result = diag.collect(root, commands=False, runtime=root / 'runtime')
        self.assertTrue(result['name'].endswith('.zip'))
        return zipfile.ZipFile(io.BytesIO(base64.b64decode(result['data'])))

    def test_stopped_service_partial_logs_and_credentials(self):
        with tempfile.TemporaryDirectory() as temp:
            root = pathlib.Path(temp)
            (root / 'logs').mkdir()
            (root / 'current').mkdir()
            (root / 'current/VERSION').write_text('0.4.3-preview.2-test')
            (root / 'logs/demo.log').write_text('password=private token=secret\nAuthorization: Bearer abc.def\n{"password":"hidden"}\nhttps://user:pass@example.org\nUseful error\n')
            (root / 'network-password.json').write_text('do-not-export')
            (root / 'state.json').write_text(json.dumps({'plugins': {'demo': {'manifest': {'name':'Demo', 'version':'1.0.0'}, 'enabled': True, 'config': {'secret':'do-not-export'}}}, 'proxy': {'http': 'do-not-export'}}))
            archive = self.archive(root)
            text = '\n'.join(archive.read(n).decode() for n in archive.namelist())
            for value in ('private', 'abc.def', 'hidden', 'user:pass', 'do-not-export'):
                self.assertNotIn(value, text)
            self.assertIn('Useful error', text)
            self.assertIn('0.4.3-preview.2-test', text)
            self.assertIn('Demo', text)
            self.assertNotIn('network-password.json', archive.namelist())

    def test_corrupt_state_and_missing_logs_still_export(self):
        with tempfile.TemporaryDirectory() as temp:
            root = pathlib.Path(temp)
            (root / 'state.json').write_text('invalid')
            archive = self.archive(root)
            self.assertTrue(json.loads(archive.read('report.json'))['warnings'])
            self.assertIn('system.json', archive.namelist())

    def test_large_logs_and_symlinks(self):
        with tempfile.TemporaryDirectory() as temp:
            root = pathlib.Path(temp)
            (root / 'logs').mkdir()
            (root / 'secret').write_text('DO_NOT_FOLLOW')
            (root / 'logs/link.log').symlink_to(root / 'secret')
            (root / 'logs/large.log').write_text('X' * (diag.MAX_FILE * 2) + '\nLAST ERROR')
            archive = self.archive(root)
            self.assertNotIn('logs/link.log', archive.namelist())
            data = archive.read('logs/large.log')
            self.assertLessEqual(len(data), diag.MAX_FILE)
            self.assertTrue(data.endswith(b'LAST ERROR'))

    def test_apk_logs_summary_are_bounded_and_exclude_application_files(self):
        with tempfile.TemporaryDirectory() as temp:
            root = pathlib.Path(temp)
            home = root / 'user'
            apk = home / '.local/share/framely/apk-manager'
            (apk / 'logs').mkdir(parents=True)
            (apk / 'apks').mkdir()
            (apk / 'apks/private.apk').write_text('PRIVATE_APK_CONTENT')
            (apk / 'state.json').write_text(json.dumps({'records': {'ctx/pkg': {
                'context':'ctx', 'steamToken':'PRIVATE_BINDING_TOKEN',
                'metadata': {'package':'test.pkg', 'name':'Test app', 'icon':'PRIVATE_ICON'},
                'removed':False, 'pending':'uninstall'}}}))
            for i in range(14):
                path = apk / 'logs' / f'{i:02}.log'
                path.write_text(f'operation {i} token=PRIVATE_LOG_TOKEN')
                import os
                os.utime(path, (i + 1, i + 1))
            saved = home / '.local/share/Steam/logs/lepton-logcats/ctx'
            saved.mkdir(parents=True)
            (saved / 'logcat-crash.log').write_text('Android boot failure password=PRIVATE_PASSWORD')
            (saved / 'video.mp4').write_text('PRIVATE_MEDIA')
            (saved / 'logcat-symlink.log').symlink_to(saved / 'video.mp4')
            result = diag.collect(root, commands=False, runtime=root/'runtime', steam_home=home)
            archive = zipfile.ZipFile(io.BytesIO(base64.b64decode(result['data'])))
            self.assertIn('apk/summary.json', archive.namelist())
            self.assertIn('apk/android-saved/ctx/logcat-crash.log', archive.namelist())
            self.assertNotIn('apk/operations/00.log', archive.namelist())
            self.assertIn('apk/operations/13.log', archive.namelist())
            text = '\n'.join(archive.read(n).decode() for n in archive.namelist())
            self.assertIn('Android boot failure', text)
            self.assertIn('Test app', text)
            self.assertNotIn('PRIVATE_', text)

    def test_active_container_diagnostics_use_session_user_and_keep_timeout_output(self):
        with tempfile.TemporaryDirectory() as temp:
            root = pathlib.Path(temp)
            (root/'steam-user').write_text('teststeam')
            home = root/'user'
            user = SimpleNamespace(pw_name='teststeam', pw_dir=str(home), pw_uid=1000)
            calls = []
            def run(args, **kwargs):
                calls.append(args)
                if 'ps' in args:
                    kwargs['stdout'].write(b'lepton-active\tUp 1 minute\nlepton-stopped\tExited (0)\n')
                elif 'exec' in args:
                    kwargs['stdout'].write(b'=== boot ===\n0\n=== user ===\nRUNNING_LOCKED\n')
                    raise subprocess.TimeoutExpired(args, 8)
                return SimpleNamespace(returncode=0)
            with mock.patch.object(diag.pwd,'getpwnam',return_value=user), mock.patch.object(diag.os,'geteuid',return_value=0), mock.patch.object(diag.subprocess,'run',side_effect=run):
                result = diag.collect(root, runtime=root/'runtime')
            archive = zipfile.ZipFile(io.BytesIO(base64.b64decode(result['data'])))
            text = archive.read('apk/android-live/lepton-active.log').decode()
            self.assertIn('RUNNING_LOCKED', text)
            executions = [a for a in calls if 'exec' in a]
            self.assertEqual(len(executions), 1)
            self.assertEqual(executions[0][:4], ['runuser','-u','teststeam','--'])
            self.assertIn('HOME='+str(home), executions[0])
            self.assertIn('XDG_RUNTIME_DIR=/run/user/1000', executions[0])
            self.assertNotIn('start', executions[0])
            self.assertTrue(any('partial output' in w for w in json.loads(archive.read('report.json'))['warnings']))

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