andgineer

from sqlalchemy import event
from sqlalchemy.orm import sessionmaker
from app.db.session import engine
import pytest
import app.tests.config


@pytest.fixture(
    scope='function',
    autouse=True  # New test DB session for each test todo we need it only for tests with Client fixture

andgineer

#! /usr/bin/env bash
#
# Check if the $1 docker-compose service is running
#
# Usage:
#   is_container_running <service name>
#

function container_is_not_running {
  if [[ -z `docker ps -q --no-trunc | grep $(docker-compose ps -q $1)` ]]; then

andgineer

for VAR in $(compgen -e); do
  # to work in zsh replace `${!VAR}` with `${(P)VAR}`
  sed -i 's|${'$VAR'}|'"${!VAR}"'|g' /config.yml
done

andgineer

for container_id in $(docker ps --filter "name=<ContainerName>" --quiet); do
 docker inspect --format='{{(index (index .NetworkSettings.Ports "<PostNumber>/tcp") 0).HostPort}}' $container_id
done

andgineer

private void loadVarsFromText(String text) {
    text.split('\n').each { envLine ->
        def (key, value) = envLine.tokenize('=')
        env."${key}" = "${value}"
    }
}

andgineer

"""
Loads a number of files listed in args:
    python expand_vars.py file1 file2 ...
Expects line format like
    key=value
Skips comments and empty lines.
Substitute vars $var / ${var} from system environment and from previous lines / files/

Universal Python 2 / 3 module to use on any Jenkins agent.
"""

andgineer

private void loadVarsFromFile(String path) {
    def file = readFile(path)
        .replaceAll("(?m)^\\s*\\r?\\n", "")  // skip empty line
        .replaceAll("(?m)^#[^\\n]*\\r?\\n", "")  // skip commented lines
    file.split('\n').each { envLine ->
        def (key, value) = envLine.tokenize('=')
        env."${key}" = "${value.trim().replaceAll('^\"|\"$', '')}"
    }
}

andgineer

sh """
	aws secretsmanager get-secret-value --secret-id accounts --query 'SecretString' --output text > secrets.json
""".stripIndent()
jsonfile = readJSON file: 'secrets.json'
env.MY_SECRET = jsonfile.MY_SECRET

andgineer

    session = get_session() if s3 is None else None
    async with session.create_client(
        "s3"
    ) if s3 is None else contextlib.AsyncExitStack() as s3_temp:
        s3_object = await (s3 or s3_temp).get_object(Bucket=bucket, Key=key)

andgineer

    session = get_session()
    async with session.create_client("s3") as s3_temp:
        s3_object = await s3_temp.get_object(Bucket=bucket, Key=key)