Тестирование облачных функций с функциями-структурами в Python

Функции Google Cloud предоставляют полностью управляемый способ развертывания простых приложений и запустить короткие запланированные операции. Как и в случае большинства вещей в Google Cloud, CF имеют столько темы для покрытия, одна статья недостаточно, чтобы даже поцарапать поверхность. Таким образом, вместо этого позвольте мне использовать этот пост для решения определенной темы CF: тестирование.

Эволюция нашей функции

Мне было поручено рефакторировать всю систему BI CO компании, которая означала создание некоторых конвейеров данных и использование API. Сначала это был один источник, запускается ежедневно с планировщиком облака через Cloud Pub/Sub. Сообщение, которое затем прочитано только «FOO», было опубликовано в паб/подразделение, которая имела функцию Беги () внутри main.py Модуль на другом конце, ожидающий, чтобы быть созданным и делать свое дело.

Необходимость ввода аргументов

Он вскоре стал очевидным, что запуск сообщений функция должна фактически содержать некоторую информацию, такую как график, который был запущен. Это было покрыто CF Docs, поэтому нет Biggie. Затем пришел еще один источник, который необходимый для рефраксина, и с ним – небольшая проблема. CF в Python Разрешить только один метод внутри модуля с именем main.py быть конечной точкой. Что означало мою Беги () Функция теперь должна была стать функцией реле, вызывая другие функции в соответствии с полезной нагрузкой сообщения. Предварительный просмотр этого виден в обложке изображения.

Потребность в средах переменных

Когда вы впервые написать свой код CF-To-Be, вы можете использовать свой код, чтобы напрямую хранить конфиденциальную информацию, например, токены и ключи API, но как только вы попадаете в этот код в репо, вам нужны другие решения. То, что я сделал, я поставил свою конфиденциальную информацию в некоторых файлах JSON и сделанный Git игнорировать их, поэтому они не будут опубликованы в репо. Но поскольку мой CF развернут от REPO GitHub компании, развернутые функции также не могли прочитать эти секреты. Таким образом, естественно, я обратился к переменным среды, которые вы можете определить при редактировании вашего ср. Они работают, как вы ожидаете, что их ценности легко доступны через Python ОС Модуль, как так:

BAMBOO_API_TOKEN = os.environ.get('BAMBOO_API_TOKEN')
BIGQUERY_PROJECT = os.environ.get('BIGQUERY_PROJECT')

Эти два требования сделали локальные тестирования безмятежными, чтобы сказать наименее.

Введите функции-фреймворки …

Модуль функций Google настраивает локальный экземпляр вашего CF, но он не работает автоматически. Вам необходимо отправить сообщение, запрос на почту с очень конкретным форматом, чтобы получить его для запуска вашего кода. Это также означало написание обработчика для сообщения, которое вызывает определенный локальный запуск вашего CF, поэтому Местный источник-1-ежедневно , Местный источник-1-еженедельный , Местный источник-2-все и т.п. Немного утомительно, но это работает так, пока он должен, если нет переменных окружающей среды. По какой-то причине вы не можете просто настроить переменные среды локально, а затем забрать их внутри CF после запуска. Решение состоит в том, чтобы генерировать файл, просто называемый .env.env. , что хранит переменные, которые вы хотите. К счастью, это также было упаковано для вас в Python.

Настройка среды

Я создал функцию, Setup-functions-fmwk.py , что принимает позиционные аргументы и генерирует .env файл.

from sys import argv
import json

# Arguments check
if len(argv) < 4:
    print("Not all arguments entered.")
    print("Args: ", ' '.join(argv))
    print("Exiting function.")
    exit()

# Some additional checks...

project = argv[1]
environment = argv[2]
runMessage = argv[3]

# Get API key
fp = 'api-key.json'
try:
    with open(fp) as f:
        j = json.load(f)
except FileNotFoundError:
    print("File `{}` not found.".format(fp))
    print("Exiting function.")
    exit()
apiKey = j['apiKey']

# Set up environment variables
with open('.env', 'w') as env:
    env.write(\n')
    env.write(\n'.format(project))
    env.write(\n'.format(apiKey))
    env.write(\n'.format(environment))
    env.write(\n'.format(runMessage))

print("Setup complete.")
exit()

Я решил передать RunMessage как EV, чтобы иметь только один локальный обработчик функции. Конечная точка выглядит так:

def run(event, context):
    '''
    Endpoint function triggered by Pub/Sub that runs a specified API.

    Args:
        event (dict) - The dictionary with data specific to this type of event.
        context (google.cloud.functions.Context) - The Cloud Functions event metadata.
    '''
    import base64
    import bambooAPI
    import pipedriveAPI
    import hnbAPI
    import gscostAPI
    from dotenv import load_dotenv
    import os

    print("This cloud function was triggered by messageId {} published at {}.".format(context.event_id, context.timestamp))

    if 'data' in event:
        pubsubMessage = base64.b64decode(event['data']).decode('utf-8')
        print("Cloud function trigerred with payload `{}`.".format(pubsubMessage))
        if pubsubMessage == 'pipedrive-legacy-run-hourly':
            print("Starting `pipedriveAPI.run_legacy_mode('hourly')`")
            pipedriveAPI.run_legacy_mode('hourly')
        # Other run options ...
        elif pubsubMessage == 'local':
            print("Loading local environment.")
            load_dotenv()
            env_run_msg = os.getenv("RUN_MESSAGE") 
            if env_run_msg == 'pipedrive-legacy-run-hourly':
                pipedriveAPI.run_legacy_mode('hourly')
            # Other run options ...
        else:
            print("Unknown payload. Terminating cloud function.")
    else:
        print("No payload in the Pub/Sub message. Terminating cloud function.")

Как вы можете видеть, однажды .evn был написан с ключ = значение Пары, это загружено просто с load_dotenv () модуль. Затем вы можете использовать ОС Как мы сделали выше, загрузить переменные.

Единственное, что осталось сделать, это отправить триггер. Документы Google объясняют, как это сделать с Curl , но это грязное утверждение, конечно, не то, что вы помните, если вы не будете тестировать свой CF 20+ раз в день, так почему бы не сделать функцию Python для создания сообщения? Я позвонил по моему Run-functions-fmwk.py :

from sys import argv
import requests
import base64
import datetime

# Set up curl call
# Check endpoint

msg = 'local'
ts = datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ")
data = base64.b64encode(msg.encode('utf-8')).decode('utf-8')
print("Encoded message: {}".format(data))

d = {
    "context": {
        "eventId":"1144231683168617",
        "timestamp":"{}".format(ts),
        "eventType":"google.pubsub.topic.publish",
        "resource": {
            "service":"pubsub.googleapis.com",
            "name":"projects/bornfight-projects/topics/gcf-test",
            "type":"type.googleapis.com/google.pubsub.v1.PubsubMessage"
        }
    },
    "data": {
        "@type": "type.googleapis.com/google.pubsub.v1.PubsubMessage",
        "attributes": {
             "attr1":"attr1-value"
        },
        "data":"{}".format(str(data))
    }
}
url = 'http://127.0.0.1:8080'
h = {"Content-Type": "application/json"}

r = requests.post(url, headers=h, json=d)

print("Response {}".format(r.status_code))

exit()

Одно следует отметить, здесь – это адрес запроса пост. По умолчанию Функциональные рамки Слушает порта 8080 у вашего localhost, поэтому, если вам нужно изменить порт, вы можете передать его в качестве аргумента для вызова функции.

Окончательный пробег

Вам нужны две терминальные сеансы, чтобы сделать это, чтобы запустить функции-фреймворки, другую отправить сообщение.

В терминале 1 запустить:

python3 setup-functions-fmwk.py bi-project test-environment pipedrive-legacy-run-hourly

Чтобы настроить среду, а затем функции вызовов:

functions-framework --target run --signature-type event

цель это название конечной точки, тип подписи это мероприятие , если только ваша функция не имеет дело с веб-кавками, то это http . , но вам также нужно изменить, как ваша функция потребляет данные. Есть некоторые другие варианты, такие как порт, который используется для прослушивания входящих запросов, но я не пойду в них.

Наконец, отправлять сообщение, перейдите к терминалу 2 и запустите:

python3 run-functions-fmwk.py

CF возвращает статус 200, если он успешно заканчивается, 500, если он сбивает, но, конечно, вам рекомендуется делать простую отладку путем печати на STDOUT, чтобы сделать вашу жизнь проще.

Функциональные рамки Не перестает работать, когда функция заканчивается, поэтому вам нужно прерывать его с помощью Ctrl + C Вернуться в терминал 1.

Мне потребовалось некоторое время, чтобы понять все это, надеюсь, это экономит вам некоторое время:)

Оригинал: “https://dev.to/bornfightcompany/testing-cloud-functions-with-functions-framework-in-python-9cf”