briggleman

import time

from contextlib import ContextDecorator



class Logger(ContextDecorator):
    """
    Logger()
    

briggleman

---
version: 1
disable_existing_loggers: False
formatters:
  simple:
    format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
  colored:
    (): "colorlog.ColoredFormatter"
    datefmt: "%Y-%m-%d %H:%M:%S"
    format: "%(white)s%(asctime)s.%(msecs)03d%(reset)s - %(cyan)s[%(module)s.%(funcName)s]%(reset)s - %(log_color)s[%(levelname)s] :=>%(reset)s %(message)s"

briggleman

import logging
import google.cloud.logging

from google.cloud.logging.handlers.transports import BackgroundThreadTransport
from google.cloud.logging.logger import _GLOBAL_RESOURCE

DEFAULT_LOGGER_NAME = "python"
EXCLUDED_LOGGER_DEFAULTS = ("google.cloud", "google.auth", "google_auth_httplib2")

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')

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

@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

@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

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

# 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

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