使用GitHub Actions和Python构建Home CI / CD

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}}'"