Строим домашний CI-CD при помощи GitHub Actions и Python

from flask import Flask
from flask import request, jsonify

app = Flask(__name__)

def validate_post_data(data: dict) -> bool:
    if not isinstance(data, dict):
        return False
    if not data.get('name') or not isinstance(data['name'], str):
        return False
    if data.get('age') and not isinstance(data['age'], int):
        return False
    return True

@app.route('/', methods=['GET'])
def hello():
    return 'Hello World!'

@app.route('/api', methods=['GET', 'POST'])
def api():
    """
    /api entpoint
    GET - returns json= {'status': 'test'}
    POST -  {
            name - str not null
            age - int optional
            }
    :return:
    """
    if request.method == 'GET':
        return jsonify({'status': 'test'})
    elif request.method == 'POST':
        if validate_post_data(request.json):
            return jsonify({'status': 'OK'})
        else:
            return jsonify({'status': 'bad input'}), 400

def main():
    app.run(host='0.0.0.0', port=8080)

if __name__ == '__main__':
    main()

import unittest
import app as tested_app
import json

class FlaskAppTests(unittest.TestCase):

    def setUp(self):
        tested_app.app.config['TESTING'] = True
        self.app = tested_app.app.test_client()

    def test_get_hello_endpoint(self):
        r = self.app.get('/')
        self.assertEqual(r.data, b'Hello World!')

    def test_post_hello_endpoint(self):
        r = self.app.post('/')
        self.assertEqual(r.status_code, 405)

    def test_get_api_endpoint(self):
        r = self.app.get('/api')
        self.assertEqual(r.json, {'status': 'test'})

    def test_correct_post_api_endpoint(self):
        r = self.app.post('/api',
                          content_type='application/json',
                          data=json.dumps({'name': 'Den', 'age': 100}))
        self.assertEqual(r.json, {'status': 'OK'})
        self.assertEqual(r.status_code, 200)

        r = self.app.post('/api',
                          content_type='application/json',
                          data=json.dumps({'name': 'Den'}))
        self.assertEqual(r.json, {'status': 'OK'})
        self.assertEqual(r.status_code, 200)

    def test_not_dict_post_api_endpoint(self):
        r = self.app.post('/api',
                          content_type='application/json',
                          data=json.dumps([{'name': 'Den'}]))
        self.assertEqual(r.json, {'status': 'bad input'})
        self.assertEqual(r.status_code, 400)

    def test_no_name_post_api_endpoint(self):
        r = self.app.post('/api',
                          content_type='application/json',
                          data=json.dumps({'age': 100}))
        self.assertEqual(r.json, {'status': 'bad input'})
        self.assertEqual(r.status_code, 400)

    def test_bad_age_post_api_endpoint(self):
        r = self.app.post('/api',
                          content_type='application/json',
                          data=json.dumps({'name': 'Den', 'age': '100'}))
        self.assertEqual(r.json, {'status': 'bad input'})
        self.assertEqual(r.status_code, 400)

if __name__ == '__main__':
    unittest.main()

name: Run tests on any Push event
# Запуск при любом push евенте в любой ветке, за исключением релизных тэгов.
# Они будт тестироваться перед сборкой
on:
  push:
    tags:
      - '!refs/tags/*'
    branches:
      - '*'
jobs:
  run_tests:
    runs-on: [ubuntu-latest]
    steps:
      # Чекаутим код
      - uses: actions/checkout@master
      # Устанавливаем python нужной версии
      - uses: actions/setup-python@v1
        with:
          python-version: '3.8'
          architecture: 'x64'
      - name: Install requirements
        # Устанавливаем зависимости
        run: pip install -r requirements.txt
      - name: Run tests
        run: coverage run src/tests.py
      - name: Tests report
        run: coverage report

# Берем нужный базовый образ
FROM python:3.8-alpine
# Копируем все файлы из текущей директории в /app контейнера
COPY ./ /app
# Устанавливаем все зависимости
RUN apk update && pip install -r /app/requirements.txt --no-cache-dir
# Устанавливаем приложение (Подробнее смотри Distutils)
RUN pip install -e /app
# Говорим контейнеру какой порт слушай
EXPOSE 8080
# Запуск нашего приложения при старте контейнера
CMD web_server

# В качестве альтернативы distutils можно просто указать что выполнить
#CMD python /app/src/app.py

# coding=utf-8
import os
import sys
import logging
import logging.config
import logging.handlers

from flask import Flask
from flask import request, jsonify
import docker

log = logging.getLogger(__name__)
app = Flask(__name__)
docker_client = docker.from_env()
MY_AUTH_TOKEN = os.getenv('CI_TOKEN', None)  # Берем наш токен из переменной окружения

def init_logging():
    """
    Инициализация логгера
    :return:
    """
    log_format = f"[%(asctime)s] [ CI/CD server ] [%(levelname)s]:%(name)s:%(message)s"
    formatters = {'basic': {'format': log_format}}
    handlers = {'stdout': {'class': 'logging.StreamHandler',
                           'formatter': 'basic'}}
    level = 'INFO'
    handlers_names = ['stdout']
    loggers = {
        '': {
            'level': level,
            'propagate': False,
            'handlers': handlers_names
        },
    }
    logging.basicConfig(level='INFO', format=log_format)
    log_config = {
        'version': 1,
        'disable_existing_loggers': False,
        'formatters': formatters,
        'handlers': handlers,
        'loggers': loggers
    }
    logging.config.dictConfig(log_config)

def get_active_containers():
    """
    Получение списка запущенных контейнеров
    :return:
    """
    containers = docker_client.containers.list()
    result = []
    for container in containers:
        result.append({
            'short_id': container.short_id,
            'container_name': container.name,
            'image_name': container.image.tags,
            'created':  container.attrs['Created'],
            'status':  container.status,
            'ports':  container.ports,
        })
    return result

def get_container_name(item: dict) -> [str, str]:
    """
    Получение имени image из POST запроса
    :param item:
    :return:
    """
    if not isinstance(item, dict):
        return ''
    owner = item.get('owner')
    repository = item.get('repository')
    tag = item.get('tag', 'latest').replace('v', '')
    if owner and repository and tag:
        return f'{owner}/{repository}:{tag}', repository
    if repository and tag:
        return f'{repository}:{tag}', repository
    return '', ''

def kill_old_container(container_name: str) -> bool:
    """
    Перед запуском нового контейнера, удаляем старый
    :param container_name:
    :return:
    """
    try:
        # Получение получение контейнера
        container = docker_client.containers.get(container_name)
        # Остановка
        container.kill()
    except Exception as e:
        # На случай если такого контейнера небыло
        log.warning(f'Error while delete container {container_name}, {e}')
        return False
    finally:
        # Удаление остановленых контейнеров, чтобы избежать конфликта имен
        log.debug(docker_client.containers.prune())
    log.info(f'Container deleted. container_name = {container_name}')
    return True

def deploy_new_container(image_name: str, container_name: str, ports: dict = None):
    try:
        # Пул последнего image из docker hub'a
        log.info(f'pull {image_name}, name={container_name}')
        docker_client.images.pull(image_name)
        log.debug('Success')
        kill_old_container(container_name)
        log.debug('Old killed')
        # Запуск нового контейнера
        docker_client.containers.run(image=image_name, name=container_name, detach=True, ports=ports)
    except Exception as e:
        log.error(f'Error while deploy container {container_name}, n{e}')
        return {'status': False, 'error': str(e)}, 400
    log.info(f'Container deployed. container_name = {container_name}')
    return {'status': True}, 200

@app.route('/', methods=['GET', 'POST'])
def MainHandler():
    """
    GET - Получение списка всех активных контейнеров
    POST - деплой сборки контейнера
    Пример тела запроса:
    {
        "owner": "gonfff",
        "repository": "ci_example",
        "tag": "v0.0.1",
         "ports": {"8080": 8080}
    }
    :return:
    """
    if request.headers.get('Authorization') != MY_AUTH_TOKEN:
        return jsonify({'message': 'Bad token'}), 401
    if request.method == 'GET':
        return jsonify(get_active_containers())
    elif request.method == 'POST':
        log.debug(f'Recieved {request.data}')
        image_name, container_name = get_container_name(request.json)
        ports = request.json.get('ports') if request.json.get('ports') else None
        result, status = deploy_new_container(image_name, container_name, ports)
        return jsonify(result), status

def main():
    init_logging()
    if not MY_AUTH_TOKEN:
        log.error('There is no auth token in env')
        sys.exit(1)
    app.run(host='0.0.0.0', port=5000)

if __name__ == '__main__':
    main()

name: Publish on Docker Hub and Deploy

on:
  release:
    types: [published]
  # Запуск только при публиковании нового релиза

jobs:
  run_tests:
    # Первую джобу смело можем копипастить из экшена для тестирования
    runs-on: [ubuntu-latest]
    steps:
      # Чекаутим код
      - uses: actions/checkout@master
      # Устанавливаем python нужной версии
      - uses: actions/setup-python@v1
        with:
          python-version: '3.8'
          architecture: 'x64'
      - name: Install requirements
        # Устанавливаем зависимости
        run: pip install -r requirements.txt
      - name: Run tests
        # Запускаем тесты
        run: coverage run src/tests.py
      - name: Tests report
        run: coverage report

  build_and_pub:
    # Если тесты были пройдены успешно
    needs: [run_tests]
    runs-on: [ubuntu-latest]
    env:
      LOGIN: ${{ secrets.DOCKER_LOGIN }}
      NAME: ${{ secrets.DOCKER_NAME }}
    steps:
      - name: Login to docker.io
        # Сначала мы логинимся в docker.io
        run:  echo ${{ secrets.DOCKER_PWD }} | docker login -u ${{ secrets.DOCKER_LOGIN }} --password-stdin
        # Чекаутим код
      - uses: actions/checkout@master
      - name: Build image
        # Собираем image и называем его так как указано в hub.docker т.е. login/repository:version
        run: docker build -t $LOGIN/$NAME:${GITHUB_REF:11} -f Dockerfile .
      - name: Push image to docker.io
        # Пушим образ в registry
        run: docker push $LOGIN/$NAME:${GITHUB_REF:11}

  deploy:
    # Если мы успешно собрали контейнер и отправили в registry, то делаем хук деплоймент серверу
    # Попробуем готовый экшен curl из маркетплэйса
    needs: [build_and_pub]
    runs-on: [ubuntu-latest]
    steps:
      - name: Set tag to env
        run: echo ::set-env name=RELEASE_VERSION::$(echo ${GITHUB_REF:11})
      - name: Send webhook for deploy
        run: "curl --silent --show-error --fail -X POST ${{ secrets.DEPLOYMENT_SERVER }} -H 'Authorization: ${{ secrets.DEPLOYMENT_TOKEN }}' -H 'Content-Type: application/json' -d '{"owner": "${{ secrets.DOCKER_LOGIN }}", "repository": "${{ secrets.DOCKER_NAME }}", "tag": "${{ env.RELEASE_VERSION }}", "ports": {"8080": 8080}}'"