tests-scan: re-enable cross-project 'job' creation

job-runner.toml

@@ -50,6 +50,7 @@ driver='github'
 [forge.github]
 clone-url = 'https://github.com/'
 api-url = 'https://api.github.com/'
+content-url = 'https://raw.githubusercontent.com/'
 post = true  # whether to post statuses, open issues, etc.
 user-agent = 'job-runner (cockpit-project/bots)'
 # (at least) one of `token` or `post = false` must be set

lib/aio/abc.py

@@ -18,14 +18,32 @@
 
 from yarl import URL
 
-from .jsonutil import JsonObject
+from .jsonutil import JsonObject, get_int, get_str
+
+
+class SubjectSpecification:
+    def __init__(self, obj: JsonObject) -> None:
+        self.repo = get_str(obj, 'repo')
+        self.sha = get_str(obj, 'sha', None)
+        self.pull = get_int(obj, 'pull', None)
+        self.branch = get_str(obj, 'branch', None)
+        self.target = get_str(obj, 'target', None)
 
 
 class Subject(NamedTuple):
-    clone_url: URL
+    forge: 'Forge'
+    repo: str
     sha: str
     rebase: str | None = None
 
+    @property
+    def clone_url(self) -> URL:
+        return self.forge.clone / self.repo
+
+    @property
+    def content_url(self) -> URL:
+        return self.forge.content / self.repo
+
 
 class Status:
     link: str
@@ -35,9 +53,10 @@ async def post(self, state: str, description: str) -> None:
 
 
 class Forge:
-    async def resolve_subject(
-        self, repo: str, sha: str | None, pull_nr: int | None, branch: str | None, target: str | None
-    ) -> Subject:
+    clone: URL
+    content: URL
+
+    async def resolve_subject(self, spec: SubjectSpecification) -> Subject:
         raise NotImplementedError
 
     async def check_pr_changed(self, repo: str, pull_nr: int, expected_sha: str) -> str | None:
@@ -49,6 +68,9 @@ def get_status(self, repo: str, sha: str, context: str | None, location: URL) ->
     async def open_issue(self, repo: str, issue: JsonObject) -> None:
         raise NotImplementedError
 
+    async def read_file(self, subject: Subject, filename: str) -> str | None:
+        raise NotImplementedError
+
     @classmethod
     def new(cls, config: JsonObject) -> Self:
         raise NotImplementedError

lib/aio/github.py

@@ -19,18 +19,35 @@
 import logging
 import platform
 from collections.abc import Mapping
-from typing import NamedTuple, Self
+from typing import Awaitable, Callable, NamedTuple, Self
 
 import aiohttp
 from yarl import URL
 
-from .abc import Forge, Status, Subject
+from .abc import Forge, Status, Subject, SubjectSpecification
 from .jsonutil import JsonError, JsonObject, JsonValue, get_bool, get_dict, get_nested, get_str, typechecked
-from .util import LRUCache, create_http_session
+from .util import LRUCache, T, create_http_session
 
 logger = logging.getLogger(__name__)
 
 
+async def retry(func: Callable[[], Awaitable[T]]) -> T:
+    for attempt in range(4):
+        try:
+            return await func()
+        except aiohttp.ClientResponseError as exc:
+            if exc.status < 500:
+                raise
+        except aiohttp.ClientError:
+            pass
+
+        # 1 → 2 → 4 → 8s delay
+        await asyncio.sleep(2 ** attempt)
+
+    # ...last attempt.
+    return await func()
+
+
 class CacheEntry(NamedTuple):
     conditions: Mapping[str, str]
     value: JsonValue
@@ -43,6 +60,7 @@ def __init__(self, config: JsonObject) -> None:
         self.config = config
         self.clone = URL(get_str(config, 'clone-url'))
         self.api = URL(get_str(config, 'api-url'))
+        self.content = URL(get_str(config, 'content-url'))
         self.dry_run = not get_bool(config, 'post')
 
     async def __aenter__(self) -> Self:
@@ -65,20 +83,7 @@ async def post_once() -> JsonValue:
                 logger.debug('response %r', response)
                 return await response.json()  # type: ignore[no-any-return]
 
-        for attempt in range(4):
-            try:
-                return await post_once()
-            except aiohttp.ClientResponseError as exc:
-                if exc.status < 500:
-                    raise
-            except aiohttp.ClientError:
-                pass
-
-            # 1 → 2 → 4 → 8s delay
-            await asyncio.sleep(2 ** attempt)
-
-        # ...last attempt.
-        return await post_once()
+        return await retry(post_once)
 
     async def get(self, resource: str) -> JsonValue:
         async def get_once() -> JsonValue:
@@ -103,20 +108,7 @@ async def get_once() -> JsonValue:
                     self.cache.add(resource, CacheEntry(conditions, value))
                     return value  # type: ignore[no-any-return]
 
-        for attempt in range(4):
-            try:
-                return await get_once()
-            except aiohttp.ClientResponseError as exc:
-                if exc.status < 500:
-                    raise
-            except aiohttp.ClientError:
-                pass
-
-            # 1 → 2 → 4 → 8s delay
-            await asyncio.sleep(2 ** attempt)
-
-        # ...last attempt.
-        return await get_once()
+        return await retry(get_once)
 
     async def get_obj(self, resource: str) -> JsonObject:
         return typechecked(await self.get(resource), dict)
@@ -140,31 +132,38 @@ async def check_pr_changed(self, repo: str, pull_nr: int, expected_sha: str) ->
     async def open_issue(self, repo: str, issue: JsonObject) -> None:
         await self.post(f'repos/{repo}/issues', issue)
 
+    async def read_file(self, subject: Subject, filename: str) -> str | None:
+        async def read_once() -> str | None:
+            try:
+                async with self.session.get(self.content / subject.repo / subject.sha / filename) as response:
+                    logger.debug('response %r', response)
+                    return await response.text()
+            except aiohttp.ClientResponseError as exc:
+                if exc.status == 404:
+                    return None
+                raise
+
+        return await retry(read_once)
+
     def get_status(self, repo: str, sha: str, context: str | None, location: URL) -> Status:
         return GitHubStatus(self, repo, sha, context, location)
 
-    async def resolve_subject(
-        self, repo: str, sha: str | None, pull_nr: int | None, branch: str | None, target: str | None
-    ) -> Subject:
-        clone_url = self.clone / repo
+    async def resolve_subject(self, spec: SubjectSpecification) -> Subject:
+        if spec.pull is not None:
+            pull = await self.get_obj(f'repos/{spec.repo}/pulls/{spec.pull}')
+            return Subject(self, spec.repo,
+                           # mypy needs some help here.  See https://github.com/python/mypy/issues/16659
+                           spec.sha if spec.sha else get_str(get_dict(pull, 'head'), 'sha'),
+                           spec.target or get_str(get_dict(pull, 'base'), 'ref'))
 
-        if sha is not None:
-            # if pull_nr is set and our sha doesn't match the PR, we will
-            # detect it soon
-            return Subject(clone_url, sha, target)
-
-        elif pull_nr is not None:
-            pull = await self.get_obj(f'repos/{repo}/pulls/{pull_nr}')
-            if target is None:
-                target = get_str(get_dict(pull, 'base'), 'ref')
-            return Subject(clone_url, get_str(get_dict(pull, 'head'), 'sha'), target)
+        elif spec.sha is not None:
+            return Subject(self, spec.repo, spec.sha, spec.target)
 
         else:
-            if not branch:
-                branch = get_str(await self.get_obj(f'repos/{repo}'), 'default_branch')
+            branch = spec.branch or get_str(await self.get_obj(f'repos/{spec.repo}'), 'default_branch')
 
-            with get_nested(await self.get_obj(f'repos/{repo}/git/refs/heads/{branch}'), 'object') as object:
-                return Subject(clone_url, get_str(object, 'sha'), target)
+            with get_nested(await self.get_obj(f'repos/{spec.repo}/git/refs/heads/{branch}'), 'object') as object:
+                return Subject(self, spec.repo, get_str(object, 'sha'), spec.target)
 
 
 class GitHubStatus(Status):

lib/aio/job.py

@@ -28,9 +28,9 @@
 from typing import Never
 
 from ..constants import BOTS_DIR
-from .abc import Forge, Subject
+from .abc import Forge, Subject, SubjectSpecification
 from .jobcontext import JobContext
-from .jsonutil import JsonObject, get_dict, get_int, get_str, get_str_map, get_strv
+from .jsonutil import JsonObject, get_dict, get_int, get_object, get_str, get_str_map, get_strv
 from .s3streamer import Index, LogStreamer
 from .spawn import run, spawn
 from .util import gather_and_cancel, read_utf8
@@ -45,14 +45,11 @@ class Failure(Exception):
 class Job:
     def __init__(self, obj: JsonObject) -> None:
         # test subject specification
-        self.repo = get_str(obj, 'repo')
-        self.sha = get_str(obj, 'sha', None)
-        self.pull = get_int(obj, 'pull', None)
-        self.branch = get_str(obj, 'branch', None)
-        self.target = get_str(obj, 'target', None)
+        self.subject = SubjectSpecification(obj)
 
         # test specification
         self.container = get_str(obj, 'container', None)
+        self.command_subject = get_object(obj, 'command-subject', SubjectSpecification, None)
         self.secrets = get_strv(obj, 'secrets', ())
         self.command = get_strv(obj, 'command', None)
         self.env = get_str_map(obj, 'env', {})
@@ -83,6 +80,14 @@ async def run_container(job: Job, subject: Subject, ctx: JobContext, log: LogStr
         cidfile = tmpdir / 'cidfile'
         attachments = tmpdir / 'attachments'
 
+        container_image = (
+            job.container or
+            await ctx.forge.read_file(subject, '.cockpit-ci/container') or
+            ctx.default_image
+        ).strip()
+
+        log.write(f'Using container image: {container_image}\n')
+
         args = [
             *ctx.container_cmd, 'run',
             # we run arbitrary commands in that container, which aren't prepared for being pid 1; reap zombies
@@ -94,7 +99,7 @@ async def run_container(job: Job, subject: Subject, ctx: JobContext, log: LogStr
             f'--env=COCKPIT_CI_LOG_URL={log.url}',
             *itertools.chain.from_iterable(args for name, args in ctx.secrets_args.items() if name in job.secrets),
 
-            job.container or ctx.default_image,
+            container_image,
 
             # we might be using podman-remote, so we can't --volume this:
             'python3', '-c', Path(f'{BOTS_DIR}/checkout-and-run').read_text(),  # lulz
@@ -150,28 +155,36 @@ async def run_container(job: Job, subject: Subject, ctx: JobContext, log: LogStr
 
 
 async def run_job(job: Job, ctx: JobContext) -> None:
-    subject = await ctx.forge.resolve_subject(job.repo, job.sha, job.pull, job.branch, job.target)
-    title = job.title or f'{job.context}@{job.repo}#{subject.sha[:12]}'
-    slug = job.slug or f'{job.repo}/{job.context or "-"}/{subject.sha[:12]}'
+    subject = await ctx.forge.resolve_subject(job.subject)
+    title = job.title or f'{job.context}@{job.subject.repo}#{subject.sha[:12]}'
+    slug = job.slug or f'{job.subject.repo}/{job.context or "-"}/{subject.sha[:12]}'
 
     async with ctx.logs.get_destination(slug) as destination:
         index = Index(destination)
         log = LogStreamer(index)
 
-        status = ctx.forge.get_status(job.repo, subject.sha, job.context, log.url)
+        status = ctx.forge.get_status(job.subject.repo, subject.sha, job.context, log.url)
         logger.info('Log: %s', log.url)
 
         try:
-            log.start(f'{title}\nRunning on: {platform.node()}\n\nJob(' + json.dumps(job.__dict__, indent=4) + ')\n')
+            log.start(
+                f'{title}\n\n'
+                f'Running on: {platform.node()}\n\n'
+                f'Job({json.dumps(job, default=lambda obj: obj.__dict__, indent=4)})\n\n'
+            )
             await status.post('pending', 'In progress')
 
-            tasks = {run_container(job, subject, ctx, log)}
+            if job.command_subject is not None:
+                command_subject = await ctx.forge.resolve_subject(job.command_subject)
+            else:
+                command_subject = subject
+            tasks = {run_container(job, command_subject, ctx, log)}
 
             if job.timeout:
                 tasks.add(timeout_minutes(job.timeout))
 
-            if job.pull:
-                tasks.add(poll_pr(ctx.forge, job.repo, job.pull, subject.sha))
+            if job.subject.pull is not None:
+                tasks.add(poll_pr(ctx.forge, job.subject.repo, job.subject.pull, subject.sha))
 
             await gather_and_cancel(tasks)
 
@@ -189,7 +202,7 @@ async def run_job(job: Job, ctx: JobContext) -> None:
                     """).lstrip(),
                     **job.report
                 }
-                await ctx.forge.open_issue(job.repo, issue)
+                await ctx.forge.open_issue(job.subject.repo, issue)
 
         except asyncio.CancelledError:
             await status.post('error', 'Cancelled')

lib/aio/jobcontext.py

@@ -68,6 +68,7 @@ def load_config(self, path: Path, name: str, *, missing_ok: bool = False) -> Non
         except FileNotFoundError as exc:
             if missing_ok:
                 logger.debug('No %s configuration found at %s', name, str(path))
+                return
             else:
                 sys.exit(f'{path}: {exc}')
         except OSError as exc:

lib/aio/jsonutil.py

@@ -83,6 +83,15 @@ def get_dict(obj: JsonObject, key: str, default: DT | _Empty = _empty) -> DT | J
     return _get(obj, lambda v: typechecked(v, dict), key, default)
 
 
+def get_object(
+    obj: JsonObject,
+    key: str,
+    constructor: Callable[[JsonObject], T],
+    default: Union[DT, _Empty] = _empty
+) -> Union[DT, T]:
+    return _get(obj, lambda v: constructor(typechecked(v, dict)), key, default)
+
+
 def get_str_map(obj: JsonObject, key: str, default: DT | _Empty = _empty) -> DT | Mapping[str, str]:
     def as_str_map(value: JsonValue) -> Mapping[str, str]:
         return {key: typechecked(value, str) for key, value in typechecked(value, dict).items()}

test/test_aio.py

@@ -10,6 +10,7 @@
 from aioresponses import CallbackResult, aioresponses
 from yarl import URL
 
+from lib.aio.abc import SubjectSpecification
 from lib.aio.github import GitHub
 from lib.aio.jobcontext import JobContext
 from lib.aio.jsonutil import JsonObject, JsonValue, json_merge_patch
@@ -20,6 +21,7 @@
 class GitHubService:
     CLONE_URL = URL('http://github.test/')
     API_URL = URL('http://api.github.test/')
+    CONTENT_URL = URL('http://content.github.test/')
     TOKEN = 'token_ABCDEFG'
     USER_AGENT = __file__  # or any magic unique string
 
@@ -33,6 +35,7 @@ def __init__(self) -> None:
         self.config: JsonObject = {
             'api-url': str(self.API_URL),
             'clone-url': str(self.CLONE_URL),
+            'content-url': str(self.CONTENT_URL),
             'post': True,
             'token': self.TOKEN,
             'user-agent': self.USER_AGENT,
@@ -219,8 +222,8 @@ async def test_github_pr_lookup(service: GitHubService, api: GitHub) -> None:
     })
 
     # Look up the sha in the PR via the REST API
-    subject = await api.resolve_subject('owner/repo', None, pull_nr, None, None)
-    assert subject == (service.CLONE_URL / repo, sha, 'main')
+    subject = await api.resolve_subject(SubjectSpecification({'repo': 'owner/repo', 'pull': pull_nr}))
+    assert subject == (api, repo, sha, 'main')
     service.assert_hits(1, 1)
 
     # The next thing that happens is that we poll this API a lot