Source code for scheduler.mock

import json

from . import AbstractSchedulerClient
from .states import JobState, TransitionError


# HACK: MockSchedulerClient is not persistent across requests
jobs = {}


[docs]class MockSchedulerClient(AbstractSchedulerClient):

[docs]    def create(self, name, image, command, **kwargs):
        """Create a new container."""
        jobs.setdefault(name, {})['state'] = JobState.created

[docs]    def destroy(self, name):
        """Destroy a container."""
        jobs.setdefault(name, {})['state'] = JobState.destroyed

[docs]    def run(self, name, image, entrypoint, command):
        """Run a one-off command."""
        # dump input into a json object for testing purposes
        return 0, json.dumps({
            'name': name,
            'image': image,
            'entrypoint': entrypoint,
            'command': command,
        })

[docs]    def start(self, name):
        """Start a container."""
        if self.state(name) not in [JobState.created,
                                    JobState.up,
                                    JobState.down,
                                    JobState.crashed,
                                    JobState.error]:
            raise TransitionError(self.state(name),
                                  JobState.up,
                                  'the container must be stopped or up to start')
        jobs.setdefault(name, {})['state'] = JobState.up

[docs]    def state(self, name):
        """Display the given job's running state."""
        return jobs.get(name, {}).get('state', JobState.initialized)

[docs]    def stop(self, name):
        """Stop a container."""
        job = jobs.get(name, {})
        if job.get('state') not in [JobState.up, JobState.crashed, JobState.error]:
            raise TransitionError(job.get('state'),
                                  JobState.up,
                                  'the container must be up to stop')
        job['state'] = JobState.down


SchedulerClient = MockSchedulerClient