kangshifu-board/test/enterprise-gogs-routes.test.py

"""Observed Gogs 0.13 route compatibility without broad write retries."""
import json
from pathlib import Path
import sys
import unittest

sys.dont_write_bytecode = True
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'deploy/enterprise-workshop'))
from bootstrap import Stopped
from gogs_setup import create_organization, inspect, lookup_organization


class OrganizationRouteTests(unittest.TestCase):
    cfg = json.loads((ROOT / 'deploy/enterprise-workshop/enterprise.json').read_bytes())

    def test_gogs013_uses_verified_admin_route_and_form_fields(self):
        calls = []
        def call(method, route, body=None, missing=False):
            calls.append((method, route, body))
            if route == '/api/v1/orgs':
                raise Stopped('GOGS_HTTP_404')
            if method == 'GET':
                return [] if route == '/api/v1/user/orgs' else None
            self.assertEqual(route, '/api/v1/admin/users/18307996893/orgs')
            self.assertEqual(body['UserName'], 'kangshifu')
            self.assertEqual(body['FullName'], '康师傅')
            return {'id': 624152, 'username': 'kangshifu'}
        api = type('Api', (), {'call': staticmethod(call)})()
        self.assertEqual(create_organization(api, self.cfg)['id'], 624152)
        self.assertEqual(len(calls), 5)

    def test_modern_route_success_does_not_try_admin_route(self):
        calls = []
        def call(method, route, body=None, missing=False):
            calls.append(route)
            self.assertEqual(body['username'], 'kangshifu')
            return {'username': 'kangshifu'}
        api = type('Api', (), {'call': staticmethod(call)})()
        create_organization(api, self.cfg)
        self.assertEqual(calls, ['/api/v1/orgs'])

    def test_authorization_or_uncertain_error_is_not_retried(self):
        for code in ('GOGS_HTTP_403', 'GOGS_HTTP_422', 'GOGS_HTTP_500', 'REQUEST_UNCERTAIN'):
            calls = []
            def call(method, route, body=None, missing=False):
                calls.append(route)
                raise Stopped(code)
            api = type('Api', (), {'call': staticmethod(call)})()
            with self.assertRaisesRegex(Stopped, code):
                create_organization(api, self.cfg)
            self.assertEqual(calls, ['/api/v1/orgs'])

    def test_existing_organization_after404_is_not_created_twice(self):
        calls = []
        def call(method, route, body=None, missing=False):
            calls.append(method)
            if method == 'POST':
                raise Stopped('GOGS_HTTP_404')
            return {'username': 'kangshifu'}
        api = type('Api', (), {'call': staticmethod(call)})()
        self.assertEqual(create_organization(api, self.cfg)['username'], 'kangshifu')
        self.assertEqual(calls, ['POST', 'GET'])

    def test_unimplemented_org_get_uses_authenticated_membership(self):
        def call(method, route, body=None, missing=False):
            if route == '/api/v1/user/orgs':
                return [{'id': 624152, 'username': 'kangshifu'}]
            return None
        api = type('Api', (), {'call': staticmethod(call)})()
        self.assertEqual(lookup_organization(api, 'kangshifu')['id'], 624152)

    def test_same_name_outside_memberships_is_never_treated_as_absent(self):
        def call(method, route, body=None, missing=False):
            if route == '/api/v1/user/orgs':
                return []
            if route == '/api/v1/users/kangshifu':
                return {'id': 624152, 'username': 'kangshifu'}
            return None
        api = type('Api', (), {'call': staticmethod(call)})()
        with self.assertRaisesRegex(Stopped, 'EXISTING_ORGANIZATION_MEMBERSHIP_UNVERIFIED'):
            lookup_organization(api, 'kangshifu')

    def test_legacy_read_reports_exists_without_claiming_owners_verified(self):
        def call(method, route, body=None, missing=False):
            self.assertEqual(method, 'GET')
            if route == '/api/v1/user':
                return {'login': '18307996893'}
            if route == '/api/v1/user/orgs':
                return [{'username': 'kangshifu'}]
            if route.startswith('/api/v1/repos/'):
                return {'private': True, 'owner': {'login': 'kangshifu'}}
            return None
        api = type('Api', (), {'call': staticmethod(call)})()
        result = inspect(api, self.cfg)
        self.assertTrue(result['organizationExists'])
        self.assertTrue(result['repositoryExists'])
        self.assertFalse(result['organizationOwnerVerified'])
        self.assertTrue(result['ownerVerificationUnavailable'])

    def test_discovered_owner_team_can_use_real_admin_members_route(self):
        def call(method, route, body=None, missing=False):
            if route == '/api/v1/user':
                return {'login': '18307996893'}
            if route == '/api/v1/orgs/kangshifu':
                return {'username': 'kangshifu'}
            if route == '/api/v1/orgs/kangshifu/teams':
                return [{'id': 61, 'name': 'Owners'}]
            if route == '/api/v1/admin/teams/61/members':
                return [{'login': '18307996893'}]
            return None
        api = type('Api', (), {'call': staticmethod(call)})()
        self.assertTrue(inspect(api, self.cfg)['organizationOwnerVerified'])


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