briggleman

# example ci/cd file for gitlab ci/cd w/ poetry
# this file will run tests on merge and will bump
# the version on commit to master using go-semrel
# go-semrel commits will trigger a final test for
# the code coverage pipeline.  this file ensures 
# both the pipeline badge and coverage badge are
# populated and that the pyproject.toml file is 
# bumped as well
stages:
  - test

briggleman

import pytest

from unittest import mock


@pytest.fixture(autouse=True)
def mock_boto():
    # change to the directory where boto3 is being called
	with mock.patch("reward.lib.aws.boto3") as mocked:
		yield mocked

briggleman

import os
import logging
import logging.config
import aiohttp_cors
​
from aiohttp import web
from tart_test.helpers import config
from tartiflette_aiohttp import register_graphql_handlers
​
​

briggleman

---
version: 1
disable_existing_loggers: False
filters:
  uuid:
    (): "clickstreamer.config.ClientUUID"
formatters:
  simple:
    format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
  colored:

briggleman

# logging filter, adds uuid to log message
class ClientUUID(logging.Filter):

    @lru_cache(maxsize=1)
    def filter(self, record):
        record.uuid = UUID
        return True

briggleman

def scoped(scope):
    """Determines if the client id sent (x-api-key) is valid and the user has the scope required to access the resource
    Args:
        scope (str): The scope required to access the resource
    """
    def wrapper(f):
        @wraps(f)
        async def decorated(request, *args, **kwargs):
            token = await get_auth_token(request)
            try:

briggleman

@pytest.fixture(autouse=True)
def today():
    return datetime(2018, 4, 17, 22, 42, 41, 717625)


@pytest.fixture(autouse=True)
def miso(today):
    # patches the datetime function in helpers.response
    # to use this patch call miso.isoformat() in the response to render the same time
    with mock.patch("sea.helpers.response.converters.datetime") as mocked:

briggleman

@pytest.fixture(autouse=True)
def redis(occasion):
    async def get(key):
        # dictionary containing compressed data
        return occasion

    # mocked main implementation of sanic_redis_ext.RedisExtension
    with mock.patch("aioredis.Redis.get") as mocked:
        mocked.side_effect = get
        yield mocked

briggleman

from unittest import mock
from box import Box


@pytest.fixture(autouse=True)
def cursor(records):
    _cursor = []

    for record in records:
        _cursor.append(Box(record))

briggleman

from unittest import mock
from sdca.models.base import PooledMySQLDatabase, MyPooledMySQLDatabase
from peewee import OperationalError


@mock.patch.object(PooledMySQLDatabase, 'commit')
@mock.patch.object(PooledMySQLDatabase, 'in_transaction')
@mock.patch.object(PooledMySQLDatabase, 'cursor')
@mock.patch.object(PooledMySQLDatabase, 'close')
@mock.patch.object(PooledMySQLDatabase, 'is_closed')