geoffkoh

class ExecutorBase(metaclass=ABCMeta):
    """ Base class for an executor """

    def __init__(self, **kwargs):
        """ Constructor """
        pass

    @abstractmethod
    def run(self, command: str) -> (str, str):
        """ Abstract method to run a command """

geoffkoh

class LocalExecutor(ExecutorBase):

    def run(self, command: str) -> (str, str):
        """ Runs the given command using subprocess """

        args = shlex.split(command)
        stdout, stderr = subprocess.Popen(args,
                                          stdout=subprocess.PIPE,
                                          stderr=subprocess.PIPE).communicate()

geoffkoh

executor = LocalExecutor()
stdout, _ = executor.run('ls -ltr')
print(stdout)

geoffkoh

class ExecutorFactory:
    """ The factory class for creating executors"""

    registry = {}
    """ Internal registry for available executors """
    
    @classmethod
    def create_executor(cls, name: str, **kwargs) -> 'ExecutorBase':
        """ Factory command to create the executor """

geoffkoh

class ExecutorFactory:
    """ The factory class for creating executors"""

    registry = {}
    """ Internal registry for available executors """

    @classmethod
    def register(cls, name: str) -> Callable:

        def inner_wrapper(wrapped_class: ExecutorBase) -> Callable:

geoffkoh

@ExecutorFactory.register('local')
class LocalExecutor(ExecutorBase):
    # The rest of the code as above
    pass

geoffkoh

local = ExecutorFactory.create_executor('local')
out, err = local.run('ls -ltra')
print(out)

geoffkoh

#!/usr/bin/env python

# Standard imports
from abc import ABCMeta, abstractmethod
import logging
import os
import shlex
import subprocess
from typing import Callable

geoffkoh

def add(a, b):
    """ Simple function to add two numbers """
    return a + b

def test_add():
    assert add(2, 4) == 6, 'The correct answer should be 6'

geoffkoh

# Standard imports 
import requests
import sqlite3

# Third party imports
import pytest

@pytest.fixture
def setup_database():
    """ Fixture to set up the in-memory database with test data """