From a5011dd94b90ce585099cd61d4a67b61118c6a08 Mon Sep 17 00:00:00 2001 From: Oleg Sukhoroslov Date: Thu, 17 Sep 2026 16:34:16 +0300 Subject: [PATCH] Add HW 2 --- homework/02-grpc-messenger/.dockerignore | 5 + homework/02-grpc-messenger/docker-compose.yml | 125 ++++++++ homework/02-grpc-messenger/readme.md | 296 ++++++++++++++++++ .../solution/client.dockerfile | 1 + .../solution/proto/messenger.proto | 5 + .../solution/server.dockerfile | 1 + .../templates/messenger-go/client.dockerfile | 11 + .../templates/messenger-go/client/go.mod | 12 + .../templates/messenger-go/client/main.go | 80 +++++ .../templates/messenger-go/proto/go.mod | 3 + .../messenger-go/proto/messenger.proto | 7 + .../templates/messenger-py/__init__.py | 0 .../messenger-py/client-async/__init__.py | 0 .../messenger-py/client-async/client.py | 60 ++++ .../client-async/requirements.txt | 3 + .../templates/messenger-py/client.dockerfile | 11 + .../templates/messenger-py/client/__init__.py | 0 .../templates/messenger-py/client/client.py | 102 ++++++ .../messenger-py/client/requirements.txt | 2 + .../templates/messenger-py/proto/__init__.py | 0 .../messenger-py/proto/messenger.proto | 5 + .../templates/messenger-py/server/__init__.py | 0 homework/02-grpc-messenger/tests/Dockerfile | 21 ++ homework/02-grpc-messenger/tests/main.py | 109 +++++++ .../02-grpc-messenger/tests/requirements.txt | 5 + .../02-grpc-messenger/tests/test_client.py | 179 +++++++++++ .../02-grpc-messenger/tests/test_proto.py | 140 +++++++++ .../02-grpc-messenger/tests/test_server.py | 285 +++++++++++++++++ 28 files changed, 1468 insertions(+) create mode 100644 homework/02-grpc-messenger/.dockerignore create mode 100644 homework/02-grpc-messenger/docker-compose.yml create mode 100644 homework/02-grpc-messenger/readme.md create mode 100644 homework/02-grpc-messenger/solution/client.dockerfile create mode 100644 homework/02-grpc-messenger/solution/proto/messenger.proto create mode 100644 homework/02-grpc-messenger/solution/server.dockerfile create mode 100644 homework/02-grpc-messenger/templates/messenger-go/client.dockerfile create mode 100644 homework/02-grpc-messenger/templates/messenger-go/client/go.mod create mode 100644 homework/02-grpc-messenger/templates/messenger-go/client/main.go create mode 100644 homework/02-grpc-messenger/templates/messenger-go/proto/go.mod create mode 100644 homework/02-grpc-messenger/templates/messenger-go/proto/messenger.proto create mode 100644 homework/02-grpc-messenger/templates/messenger-py/__init__.py create mode 100644 homework/02-grpc-messenger/templates/messenger-py/client-async/__init__.py create mode 100644 homework/02-grpc-messenger/templates/messenger-py/client-async/client.py create mode 100644 homework/02-grpc-messenger/templates/messenger-py/client-async/requirements.txt create mode 100644 homework/02-grpc-messenger/templates/messenger-py/client.dockerfile create mode 100644 homework/02-grpc-messenger/templates/messenger-py/client/__init__.py create mode 100644 homework/02-grpc-messenger/templates/messenger-py/client/client.py create mode 100644 homework/02-grpc-messenger/templates/messenger-py/client/requirements.txt create mode 100644 homework/02-grpc-messenger/templates/messenger-py/proto/__init__.py create mode 100644 homework/02-grpc-messenger/templates/messenger-py/proto/messenger.proto create mode 100644 homework/02-grpc-messenger/templates/messenger-py/server/__init__.py create mode 100644 homework/02-grpc-messenger/tests/Dockerfile create mode 100644 homework/02-grpc-messenger/tests/main.py create mode 100644 homework/02-grpc-messenger/tests/requirements.txt create mode 100644 homework/02-grpc-messenger/tests/test_client.py create mode 100644 homework/02-grpc-messenger/tests/test_proto.py create mode 100644 homework/02-grpc-messenger/tests/test_server.py diff --git a/homework/02-grpc-messenger/.dockerignore b/homework/02-grpc-messenger/.dockerignore new file mode 100644 index 0000000..d291891 --- /dev/null +++ b/homework/02-grpc-messenger/.dockerignore @@ -0,0 +1,5 @@ +target +**/target +**/__pycache__ +**/.pytest_cache +*.pyc diff --git a/homework/02-grpc-messenger/docker-compose.yml b/homework/02-grpc-messenger/docker-compose.yml new file mode 100644 index 0000000..1889104 --- /dev/null +++ b/homework/02-grpc-messenger/docker-compose.yml @@ -0,0 +1,125 @@ +x-tests-image: &tests-image + image: grpc-solution-tests + build: + context: . + dockerfile: tests/Dockerfile + +services: + server: + image: grpc-solution-server + build: + context: ./solution + dockerfile: server.dockerfile + environment: + MESSENGER_SERVER_PORT: 51075 + ports: + - "${MESSENGER_SERVER_HOST_PORT:-51075}:51075" + # Start the clients first to exercise their reconnect logic during manual runs. + depends_on: + - client1 + - client2 + + server-tests: + image: grpc-solution-server + build: + context: ./solution + dockerfile: server.dockerfile + environment: + MESSENGER_SERVER_PORT: 51075 + + client1: + image: grpc-solution-client + build: + context: ./solution + dockerfile: client.dockerfile + environment: + MESSENGER_SERVER_ADDR: server:51075 + MESSENGER_HTTP_PORT: 8080 + ports: + - "${MESSENGER_CLIENT1_HOST_PORT:-8080}:8080" + + client2: + image: grpc-solution-client + build: + context: ./solution + dockerfile: client.dockerfile + environment: + MESSENGER_SERVER_ADDR: server:51075 + MESSENGER_HTTP_PORT: 8080 + ports: + - "${MESSENGER_CLIENT2_HOST_PORT:-8081}:8080" + + client-test-server: + image: "${MESSENGER_CLIENT_TEST_SERVER_IMAGE:-distsys.ru/course/grpc-messenger:latest}" + entrypoint: + - /usr/local/bin/client-test-server + environment: + MESSENGER_CLIENT_TEST_SERVER_EXPECTED_STREAMS: 2 + MESSENGER_CLIENT_TEST_SERVER_PROTO: /submission/proto/messenger.proto + MESSENGER_SERVER_PORT: 51075 + volumes: + - type: bind + source: "${MESSENGER_CLIENT_TEST_SERVER_PROTO_DIR:-./solution/proto}" + target: /submission/proto + read_only: true + user: "65532:65532" + read_only: true + cap_drop: + - ALL + security_opt: + - no-new-privileges:true + + client-test1: + image: grpc-solution-client + build: + context: ./solution + dockerfile: client.dockerfile + environment: + MESSENGER_SERVER_ADDR: client-test-server:51075 + MESSENGER_HTTP_PORT: 8080 + depends_on: + - client-test-server + + client-test2: + image: grpc-solution-client + build: + context: ./solution + dockerfile: client.dockerfile + environment: + MESSENGER_SERVER_ADDR: client-test-server:51075 + MESSENGER_HTTP_PORT: 8080 + depends_on: + - client-test-server + + tests: + <<: *tests-image + environment: + MESSENGER_TEST_CLIENT1_ADDR: client-test1:8080 + MESSENGER_TEST_CLIENT2_ADDR: client-test2:8080 + MESSENGER_TEST_SERVER_ADDR: server-tests:51075 + depends_on: + - server-tests + - client-test1 + - client-test2 + + server-test-runner: + <<: *tests-image + command: + - --component + - server + environment: + MESSENGER_TEST_SERVER_ADDR: server-tests:51075 + depends_on: + - server-tests + + client-tests: + <<: *tests-image + command: + - --component + - client + environment: + MESSENGER_TEST_CLIENT1_ADDR: client-test1:8080 + MESSENGER_TEST_CLIENT2_ADDR: client-test2:8080 + depends_on: + - client-test1 + - client-test2 diff --git a/homework/02-grpc-messenger/readme.md b/homework/02-grpc-messenger/readme.md new file mode 100644 index 0000000..213779e --- /dev/null +++ b/homework/02-grpc-messenger/readme.md @@ -0,0 +1,296 @@ +# Мессенджер на gRPC + +Опишите gRPC-интерфейс и реализуйте **сервер** и **клиент** мессенджера с одним общим чатом. Сервер и клиент общаются по gRPC, пользователь обращается к клиенту по HTTP. + +У сервера два RPC-метода: `SendMessage` отправляет сообщение в чат, `ReadMessages` открывает подписку на новые сообщения. Сервер должен обрабатывать несколько запросов одновременно, в том числе принимать сообщения при открытых подписках. + +Клиент при запуске открывает подписку и сохраняет сообщения от сервера в буфере в порядке получения. Через HTTP пользователь отправляет сообщения и забирает содержимое буфера. + +На схеме показаны сервер и два клиента. Тесты обращаются к клиентам от имени двух пользователей: + +```mermaid +flowchart RL + subgraph Tests + U1{User 1} + U2{User 2} + end + + subgraph Clients with HTTP interface + C1(fa:fa-comments Client 1) + C2(fa:fa-comments Client 2) + end + + subgraph gRPC server + S(fa:fa-server Server) + end + + + C1 -- SendMessage --> S + S -. Stream ReadMessages .-> C1 + + U1 -- POST /sendMessage --> C1 + C1 -- Forward messages in /getAndFlushMessages --> U1 + + C2 -- SendMessage --> S + S -. Stream ReadMessages .-> C2 + + U2 -- POST /sendMessage --> C2 + C2 -- Forward messages in /getAndFlushMessages --> U2 +``` + +## Требования + +### Доставка сообщений + +- Подписка начинается, когда сервер регистрирует вызов `ReadMessages`, и действует до отмены RPC или закрытия соединения. Сообщения, принятые до регистрации, в подписку не попадают. +- Сервер передаёт каждое сообщение ровно один раз во все подписки, активные в момент его принятия, включая подписку отправителя. Восстанавливать историю после разрывов и перезапусков не нужно. +- Сообщения, общие для нескольких подписчиков, должны идти в одинаковом порядке во всех их потоках. Клиенты сохраняют этот порядок. Для одновременных вызовов `SendMessage` сервер может выбрать любой порядок. +- `sendTime` — серверное время принятия сообщения. В течение одного запуска эти значения должны быть уникальны и строго возрастать в порядке рассылки. Ответ `SendMessage` и сообщение во всех подписках содержат одинаковый `sendTime`. +- Успешный ответ `SendMessage` означает, что сервер принял сообщение. Это не гарантирует, что все клиенты уже его получили. + +### HTTP-интерфейс клиента + +Пользователи и тесты обращаются к клиенту через два HTTP-метода. + +При успехе оба метода возвращают HTTP `200` и JSON. Поле `sendTime` — строка в [JSON-формате `google.protobuf.Timestamp`](https://protobuf.dev/reference/protobuf/google.protobuf/#timestamp), например `2025-09-20T10:58:42.665193557Z`. + +``` +POST /sendMessage +Отправляет одно сообщение в общий чат. + +Тело запроса: +{ + "author": "Ivan Ivanov", + "text": "Hey guys" +} + +Тело ответа: +{ + "sendTime": "..." +} +``` +``` +POST /getAndFlushMessages +Возвращает накопленные сообщения в порядке получения и очищает буфер. + +Тело запроса: нет + +Тело ответа: +[{ + "author": "Ivan Ivanov", + "text": "Hey guys", + "sendTime": "..." +},{ + "author": "Petr Petrov", + "text": "Hey Ivan", + "sendTime": "..." +}] +``` + +Если буфер пуст, `getAndFlushMessages` сразу возвращает `[]`. Чтение и очистка буфера должны быть атомарными: сообщение, пришедшее во время этой операции, попадает в текущий или следующий ответ. Клиент не должен терять сообщения, выдавать их повторно или менять их порядок. + +### gRPC-интерфейс сервера + +Тесты проверяют сервер отдельно от клиента. Соблюдайте требования к интерфейсу: + +- синтаксис — `proto3`, пакет — `mes_grpc`; +- gRPC-сервис `MessengerServer` содержит два метода: `SendMessage` и `ReadMessages`; +- `SendMessage` — унарный вызов. Запрос содержит одиночные строковые поля `author` и `text`, ответ — одиночное поле `sendTime` типа `google.protobuf.Timestamp`; +- `ReadMessages` принимает один пустой запрос и возвращает поток сообщений. Можно описать свой тип пустого сообщения или взять готовый из библиотеки. Каждое сообщение в потоке содержит одиночные поля `author` и `text` типа `string` и `sendTime` типа `google.protobuf.Timestamp`. + +Все перечисленные поля одного сообщения должны допускать одновременное заполнение. + +Имена типов сообщений и номера полей выберите самостоятельно — тесты их не фиксируют. + +## Оценивание + +За задание можно получить 10 баллов: + +- **2 балла** — протокол `messenger.proto`, проверяется в `test_proto.py`. +- **4 балла** — сервер, проверяется в `test_server.py`. +- **4 балла** — клиент, проверяется в `test_client.py`. + +В отчёте `solution/readme.md` опишите структуру решения, какие компоненты вы реализовали и как работают методы сервера и клиента. Без отчёта тесты запускаются, но защита не проводится и решение не засчитывается — см. [общие правила сдачи](../readme.md#сдача-решения). + +Баллы за компонент начисляются, только если прошли все его тесты: 2 или 0 за протокол, 4 или 0 за сервер, 4 или 0 за клиент. Значение `SCORE` в выводе тестов — предварительная оценка. Итоговую оценку преподаватель выставляет после защиты с учётом штрафов ниже. + +Система собирает и проверяет сервер и клиент независимо. Если сервер не собирается или не запускается, он получает 0 баллов, но клиент всё равно проверяется, и наоборот. Протокол проверяется отдельно. + +При независимом оценивании клиенты работают со служебным gRPC-сервером, построенным по вашему `messenger.proto`. Поэтому для проверки клиента нужен корректный протокол, но ошибки вашего сервера не влияют на баллы за клиент. + +На защите можно потерять баллы за следующие ошибки: + +- Сервер не может обрабатывать несколько запросов одновременно — 2 балла. +- При конкурентном доступе сервер может потерять или продублировать сообщения, выдать их в разном порядке в потоках `ReadMessages` либо нарушить требования к `sendTime` — 2 балла. +- Клиент теряет, повторно выдаёт или меняет порядок сообщений из потока `ReadMessages` — 2 балла. + +За ошибки сервера снимаются только баллы за сервер, за ошибки клиента — только баллы за клиент, не больше 4 баллов в каждом случае. Баллы за протокол сохраняются. На защите нужно разобрать предложенный преподавателем сценарий конкурентного выполнения и объяснить по своему коду, почему решение работает правильно. + +## Заготовки для клиента + +В `templates` есть заготовки клиента на Python, Python с asyncio и Go. Официальная заготовка — `messenger-py/client`; она проверена для текущего задания. Остальные заготовки относятся к прошлым версиям задания: их можно использовать, но расхождения с условием нужно исправить самостоятельно. Можно выбрать и другой язык — тесты обращаются к решению через HTTP и gRPC. + +## Порядок выполнения задания + +Выполняйте команды из папки `homework/02-grpc-messenger`. Примеры с переменными окружения написаны для Bash. В Windows используйте WSL или задавайте переменные через PowerShell. + +### Подготовка окружения + +Установите Python 3.12 или новее и Docker по [общей инструкции](../readme.md#настройка-окружения). Для Python-заготовки и локального запуска тестов установите зависимости: + +```bash +python3 -m pip install -r templates/messenger-py/client/requirements.txt -r tests/requirements.txt +grpcurl -version +``` + +Если `grpcurl` не найден, установите его по инструкции в разделе «Полезные материалы». На Windows используйте `python` вместо `python3`. + +### Структура проекта + +Разместите решение в папке `solution`. Сохраните пути к трём файлам, которые используют тесты и [docker-compose.yml](docker-compose.yml): + +- `client.dockerfile` — сборка и запуск клиента; +- `server.dockerfile` — сборка и запуск сервера; +- `proto/messenger.proto` — описание gRPC-интерфейса; дополните начальный файл. + +При сдаче отправляется только папка `solution`. Изменения за её пределами не учитываются. + +Для официальной Python-заготовки скопируйте `templates/messenger-py/client` в `solution/client`, а образец `client.dockerfile` — в `solution/client.dockerfile`. Сервер разместите в `solution/server/server.py`. Заготовка использует пакет `solution` и импорты `from solution.proto import messenger_pb2, messenger_pb2_grpc`. В своей реализации можно выбрать другую структуру, сохранив три обязательных пути выше. + +### Описание и компиляция gRPC-интерфейса + +Опишите сообщения и сервис в `solution/proto/messenger.proto`. + +Сгенерируйте код для выбранного языка с помощью `protoc`: + +```bash +# Python +python3 -m grpc_tools.protoc -I. --python_out=. --pyi_out=. --grpc_python_out=. solution/proto/messenger.proto + +# Go (после установки protoc и плагинов protoc-gen-go и protoc-gen-go-grpc) +protoc -I solution/proto --go_out=solution/proto --go_opt=paths=source_relative --go-grpc_out=solution/proto --go-grpc_opt=paths=source_relative messenger.proto +``` + +Для Go укажите в `option go_package` путь пакета в вашем модуле; пример есть в Go-заготовке. Закрепите версии генераторов, совместимые с вашей версией Go. Сгенерированные файлы включите в решение или генерируйте при сборке образа с закреплёнными версиями инструментов. + +### Реализация сервера и клиента + +Сервер реализуйте с нуля; заготовки для него нет. Можно использовать потоки или асинхронный код. Открытые подписки не должны мешать обработке других запросов. + +Если используете Python-заготовку клиента, заполните места с пометкой TODO. HTTP-сервер в ней уже реализован. При запуске клиент должен дождаться сервера, открыть `ReadMessages` и принимать сообщения независимо от обработки HTTP-запросов. + +Сервер и клиент должны брать настройки из переменных окружения: + +| Переменная | Назначение | +| --- | --- | +| `MESSENGER_SERVER_PORT` | Порт gRPC-сервера, по умолчанию `51075` | +| `MESSENGER_SERVER_ADDR` | Адрес gRPC-сервера для подключения клиента | +| `MESSENGER_HTTP_PORT` | Порт HTTP-интерфейса клиента | + +Сервер и HTTP-интерфейс клиента должны слушать на `0.0.0.0`. + +В `solution/server.dockerfile` и `solution/client.dockerfile` опишите сборку и запуск сервера и клиента. Контекст сборки — папка `solution`. Включите в образы все нужные файлы и зависимости. За образец можно взять Dockerfile заготовки. + +## Тестирование решения + +Публичные тесты проверяют [протокол](tests/test_proto.py), [сервер](tests/test_server.py) и [клиент](tests/test_client.py). Сложные конкурентные сценарии вы разберёте на защите. + +### Полная проверка + +Рекомендуемый запуск в окружении тестирующей системы: + +```bash +docker run --privileged --pull always --rm -v ./solution:/hw/solution distsys.ru/course/grpc-messenger:latest +``` + +Сервер и клиент собираются и проверяются независимо: ошибка сборки одного не мешает проверить другой. + +### Отдельные компоненты через Docker Compose + +```bash +docker compose build tests + +# Протокол +docker compose run --rm --no-deps tests --component proto + +# Сервер +docker compose build server-tests +docker compose run --rm server-test-runner + +# Клиент со служебным сервером +docker compose build client-test1 +docker compose run --rm client-tests +``` + +После изменения кода пересоберите соответствующий образ и повторите проверку. После изменения протокола заново сгенерируйте код и выполните: + +```bash +docker compose down +docker compose build tests server-tests client-test1 +docker compose run --rm tests +``` + +Смотрите логи через `docker compose logs`, останавливайте контейнеры командой `docker compose down`. После обновления задания скачайте свежий служебный образ: `docker compose pull client-test-server`. + +### Ручная отладка своей связки (необязательно) + +Запустите свой сервер и два клиента через Compose: + +```bash +docker compose build server client1 +docker compose up -d server client1 client2 +``` + +Сервер доступен на `localhost:51075`, клиенты — на `localhost:8080` и `localhost:8081`. Compose запускает клиентов раньше сервера. После правок пересоберите соответствующий образ и повторите `up`; для логов и остановки используйте команды выше. + +Без Docker запустите компоненты и тестер в отдельных терминалах из папки задания. Для официальной Python-заготовки и сервера в `solution/server/server.py`: + +```bash +# Терминал 1 +python3 -m solution.server.server +# Терминал 2 +python3 -m solution.client.client +# Терминал 3 +MESSENGER_HTTP_PORT=8081 python3 -m solution.client.client +# Терминал 4 +python3 tests/main.py +``` + +Если вы добавили зависимости, установите и их. Чтобы проверить один компонент, передайте тестеру `--component proto`, `--component server` или `--component client`. По умолчанию проверяются все компоненты (`--component all`). + +Здесь клиенты работают с вашим сервером, поэтому его ошибки могут повлиять на клиентские тесты. Для независимой проверки клиента используйте Compose со служебным сервером. + +Примеры HTTP-запросов к запущенному клиенту: + +```bash +curl -X POST localhost:8080/sendMessage -d '{"author": "alice", "text": "hey"}' +curl -X POST localhost:8080/getAndFlushMessages +``` + +### Сдача решения + +Подготовьте `solution/readme.md` и отправьте решение по [общей инструкции](../readme.md#сдача-решения). В журнале проверки будут вывод сборки, результаты тестов (после строки `=== RUN TESTS`) и логи контейнеров. + +## Полезные материалы + +### grpcurl + +[grpcurl](https://github.com/fullstorydev/grpcurl) позволяет вызывать gRPC-методы из терминала. Для Linux и Windows скачайте архив для своей ОС и архитектуры со [страницы релизов](https://github.com/fullstorydev/grpcurl/releases), распакуйте его и добавьте каталог с исполняемым файлом в `PATH`. В macOS: `brew install grpcurl`. + +Для проверки сервера откройте подписку в одном терминале, а в другом отправьте сообщение: + +```bash +# Терминал 1: поток остаётся открытым; Ctrl+C отменяет вызов +grpcurl -proto solution/proto/messenger.proto -plaintext localhost:51075 mes_grpc.MessengerServer/ReadMessages + +# Терминал 2 +grpcurl -proto solution/proto/messenger.proto -d '{"author": "alice", "text": "hello"}' -plaintext localhost:51075 mes_grpc.MessengerServer/SendMessage +``` + +### Конкурентная обработка в Python + +При конкурентном доступе к общим изменяемым данным учитывайте возможные гонки. Структуры данных и способы синхронизации выберите самостоятельно. + +- [gRPC Basics Tutorial](https://grpc.io/docs/languages/python/basics/) и [официальные примеры](https://github.com/grpc/grpc/blob/master/examples). +- Документация Python: [`threading`](https://docs.python.org/3/library/threading.html), [`queue`](https://docs.python.org/3/library/queue.html), [`asyncio`](https://docs.python.org/3/library/asyncio.html). +- Для работы с несколькими терминалами при желании можно использовать [tmux](https://github.com/tmux/tmux/wiki/Getting-Started). diff --git a/homework/02-grpc-messenger/solution/client.dockerfile b/homework/02-grpc-messenger/solution/client.dockerfile new file mode 100644 index 0000000..18b899f --- /dev/null +++ b/homework/02-grpc-messenger/solution/client.dockerfile @@ -0,0 +1 @@ +# TODO: Write Docker file for client \ No newline at end of file diff --git a/homework/02-grpc-messenger/solution/proto/messenger.proto b/homework/02-grpc-messenger/solution/proto/messenger.proto new file mode 100644 index 0000000..40ca766 --- /dev/null +++ b/homework/02-grpc-messenger/solution/proto/messenger.proto @@ -0,0 +1,5 @@ +syntax = "proto3"; + +package mes_grpc; + +// TODO: Add messages and service \ No newline at end of file diff --git a/homework/02-grpc-messenger/solution/server.dockerfile b/homework/02-grpc-messenger/solution/server.dockerfile new file mode 100644 index 0000000..0569d01 --- /dev/null +++ b/homework/02-grpc-messenger/solution/server.dockerfile @@ -0,0 +1 @@ +# TODO: Write Docker file for server \ No newline at end of file diff --git a/homework/02-grpc-messenger/templates/messenger-go/client.dockerfile b/homework/02-grpc-messenger/templates/messenger-go/client.dockerfile new file mode 100644 index 0000000..e6f24b4 --- /dev/null +++ b/homework/02-grpc-messenger/templates/messenger-go/client.dockerfile @@ -0,0 +1,11 @@ +FROM golang:1.23-alpine AS builder + +WORKDIR /grpc-messenger +COPY proto proto +COPY client client +RUN cd client && go mod download && go build . + +FROM alpine:latest +WORKDIR /grpc-messenger +COPY --from=builder /grpc-messenger . +CMD ["./client/client"] diff --git a/homework/02-grpc-messenger/templates/messenger-go/client/go.mod b/homework/02-grpc-messenger/templates/messenger-go/client/go.mod new file mode 100644 index 0000000..627cfd9 --- /dev/null +++ b/homework/02-grpc-messenger/templates/messenger-go/client/go.mod @@ -0,0 +1,12 @@ +module github.com/distsys-course/grpc-messenger/client + +go 1.23 + +replace github.com/distsys-course/grpc-messenger/grpc => ../proto + +require ( + github.com/distsys-course/grpc-messenger/grpc v0.0.0-00010101000000-000000000000 + github.com/gin-gonic/gin v1.10.1 + github.com/golang/protobuf v1.5.4 + google.golang.org/grpc v1.75.0 +) diff --git a/homework/02-grpc-messenger/templates/messenger-go/client/main.go b/homework/02-grpc-messenger/templates/messenger-go/client/main.go new file mode 100644 index 0000000..ddd2fce --- /dev/null +++ b/homework/02-grpc-messenger/templates/messenger-go/client/main.go @@ -0,0 +1,80 @@ +package main + +import ( + "context" + "fmt" + "net/http" + "os" + "sync" + "time" + + mes_grpc "github.com/distsys-course/grpc-messenger/grpc" + "github.com/gin-gonic/gin" + "github.com/golang/protobuf/jsonpb" + "google.golang.org/grpc" +) + +type ChatMessage struct { + Author string `json:"author"` + Text string `json:"text"` + SendTime time.Time `json:"sendTime"` +} + +type MessengerClient struct { + pendingMessages []ChatMessage + pendingMutex sync.Mutex + grpcClient YourMessengerServerClient +} + +func NewMessengerClient(serverAddr string) *MessengerClient { + // TODO +} + +func (c *MessengerClient) ReadMessages() { + // TODO: implement messages consumer here +} + +func (c *MessengerClient) GetPending() (messages []ChatMessage) { + c.pendingMutex.Lock() + result := c.pendingMessages + c.pendingMessages = nil + c.pendingMutex.Unlock() + return result +} + +type MessageResponse struct { + SendTime *time.Time `json:"sendTime"` + Error *string `json:"error"` +} + +func main() { + r := gin.Default() + serverAddr := os.Getenv("MESSENGER_SERVER_ADDR") + if serverAddr == "" { + serverAddr = "localhost:51075" + fmt.Println("Missing MESSENGER_SERVER_ADDR variable, using default value: " + serverAddr) + } + // TODO: create your grpc client with given address + r.POST("/getAndFlushMessages", func(c *gin.Context) { + c.JSON(http.StatusOK, client.GetPending()) + }) + + r.POST("/sendMessage", func(c *gin.Context) { + // TODO: implement send message here, that parses body into protobuf and sends to the server + c.JSON(http.StatusOK, MessageResponse{SendTime: nil}) // TODO: do not forget to fill SendTime + return + }) + + // TODO: run consumer in a goroutine + + addr := os.Getenv("MESSENGER_HTTP_PORT") + if addr == "" { + addr = "0.0.0.0:8080" + fmt.Println("Missing MESSENGER_HTTP_PORT variable, using default value: 8080") + } else { + addr = "0.0.0.0:" + addr + } + if err := r.Run(addr); err != nil { + panic(err) + } +} diff --git a/homework/02-grpc-messenger/templates/messenger-go/proto/go.mod b/homework/02-grpc-messenger/templates/messenger-go/proto/go.mod new file mode 100644 index 0000000..6066ce4 --- /dev/null +++ b/homework/02-grpc-messenger/templates/messenger-go/proto/go.mod @@ -0,0 +1,3 @@ +module github.com/distsys-course/grpc-messenger/grpc + +go 1.23 diff --git a/homework/02-grpc-messenger/templates/messenger-go/proto/messenger.proto b/homework/02-grpc-messenger/templates/messenger-go/proto/messenger.proto new file mode 100644 index 0000000..791e5a0 --- /dev/null +++ b/homework/02-grpc-messenger/templates/messenger-go/proto/messenger.proto @@ -0,0 +1,7 @@ +syntax = "proto3"; + +package mes_grpc; + +option go_package = "proto/;mes_grpc"; + +// TODO: Add messages and service \ No newline at end of file diff --git a/homework/02-grpc-messenger/templates/messenger-py/__init__.py b/homework/02-grpc-messenger/templates/messenger-py/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/homework/02-grpc-messenger/templates/messenger-py/client-async/__init__.py b/homework/02-grpc-messenger/templates/messenger-py/client-async/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/homework/02-grpc-messenger/templates/messenger-py/client-async/client.py b/homework/02-grpc-messenger/templates/messenger-py/client-async/client.py new file mode 100644 index 0000000..cb8c4ce --- /dev/null +++ b/homework/02-grpc-messenger/templates/messenger-py/client-async/client.py @@ -0,0 +1,60 @@ +import asyncio +import copy +import json +import os +from typing import List + +from aiohttp import web + + +# TODO: implement grpc client for messenger service + +class MessengerHandler: + _pendingMessages: List[dict] # list of messages, that have not been requested yet via get_messages + _pendingMessagesLock: asyncio.Lock + _grpcClient = None # grpc client of the messenger service + + def __init__(self): + self._pendingMessages = [] + self._pendingMessagesLock = asyncio.Lock() + + async def send_message(self, request): + """ + Body should be of the form: + {"author": "Ivan", "text": "hey guys"} + :return web.json_response of the form {"sendTime": ... } + """ + j = await request.json() # TODO: use google.protobuf.json_format.ParseDict and raise BadRequest on error + if 'author' not in j or 'text' not in j: + raise web.HTTPBadRequest + print('Got message to send:', json.dumps(j)) + + # TODO: your rpc call of the messenger here + + raise NotImplementedError + return web.json_response({'sendTime': ""}) # TODO: use google.protobuf.json_format.MessageToDict here + + async def get_messages(self, _): + async with self._pendingMessagesLock: + res: List[dict] = copy.deepcopy(self._pendingMessages) + self._pendingMessages = [] + return web.json_response(res) + + # TODO: implement message stream consumer in async method, that fills self._pendingMessages + # btw, consumption can be lazy and happen on get_messages, implement in any suitable way + + +if __name__ == '__main__': + app = web.Application() + grpcServerAddr = os.environ.get('MESSENGER_SERVER_ADDR', 'localhost:51075') + + # TODO: create your grpc client with given address and pass it to MessengerHandler constructor + + handler = MessengerHandler() + app.add_routes([web.post('/getAndFlushMessages', handler.get_messages)]) + app.add_routes([web.post('/sendMessage', handler.send_message)]) + + # TODO: run message stream consumer in a background coroutine + + httpPort = os.environ.get('MESSENGER_HTTP_PORT', '8080') + web.run_app(app, host='0.0.0.0', port=httpPort) diff --git a/homework/02-grpc-messenger/templates/messenger-py/client-async/requirements.txt b/homework/02-grpc-messenger/templates/messenger-py/client-async/requirements.txt new file mode 100644 index 0000000..ec60284 --- /dev/null +++ b/homework/02-grpc-messenger/templates/messenger-py/client-async/requirements.txt @@ -0,0 +1,3 @@ +aiohttp==3.12.15 +grpcio==1.75.0 +grpcio-tools==1.75.0 \ No newline at end of file diff --git a/homework/02-grpc-messenger/templates/messenger-py/client.dockerfile b/homework/02-grpc-messenger/templates/messenger-py/client.dockerfile new file mode 100644 index 0000000..914c93b --- /dev/null +++ b/homework/02-grpc-messenger/templates/messenger-py/client.dockerfile @@ -0,0 +1,11 @@ +FROM python:3.12-slim + +WORKDIR /grpc-messenger + +COPY client/requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY client/client.py solution/client/ +COPY proto solution/proto/ + +ENTRYPOINT ["python", "-m", "solution.client.client"] diff --git a/homework/02-grpc-messenger/templates/messenger-py/client/__init__.py b/homework/02-grpc-messenger/templates/messenger-py/client/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/homework/02-grpc-messenger/templates/messenger-py/client/client.py b/homework/02-grpc-messenger/templates/messenger-py/client/client.py new file mode 100644 index 0000000..cf60b80 --- /dev/null +++ b/homework/02-grpc-messenger/templates/messenger-py/client/client.py @@ -0,0 +1,102 @@ +import copy +import json +import os +import threading +from http import HTTPStatus +from http.server import HTTPServer, BaseHTTPRequestHandler +from typing import List, Dict + +import google.protobuf.empty_pb2 # Empty +import google.protobuf.json_format # ParseDict, MessageToDict +import grpc + +from solution.proto import messenger_pb2 +from solution.proto import messenger_pb2_grpc + + +class PostBox: + def __init__(self): + self._messages: List[Dict] = [] + self._lock = threading.Lock() + + def collect_messages(self) -> List[Dict]: + with self._lock: + messages = copy.deepcopy(self._messages) + self._messages = [] + return messages + + def put_message(self, message: Dict): + with self._lock: + self._messages.append(message) + + +class MessageHandler(BaseHTTPRequestHandler): + _stub = None + _postbox: PostBox + + def _read_content(self): + content_length = int(self.headers['Content-Length']) + bytes_content = self.rfile.read(content_length) + return bytes_content.decode('ascii') + + # noinspection PyPep8Naming + def do_POST(self): + if self.path == '/sendMessage': + response = self._send_message(self._read_content()) + elif self.path == '/getAndFlushMessages': + response = self._get_messages() + else: + self.send_error(HTTPStatus.NOT_IMPLEMENTED) + self.end_headers() + return + + response_bytes = json.dumps(response).encode('ascii') + self.send_response(HTTPStatus.OK) + self.send_header('Content-Length', str(len(response_bytes))) + self.end_headers() + self.wfile.write(response_bytes) + + def _send_message(self, content: str) -> dict: + json_request = json.loads(content) + + # TODO: use google.protobuf.json_format.ParseDict + + # TODO: your rpc call of the messenger here + + # TODO: use google.protobuf.json_format.MessageToDict here + return {'sendTime': ''} + + def _get_messages(self) -> List[dict]: + return self._postbox.collect_messages() + + +def main(): + grpc_server_address = os.environ.get('MESSENGER_SERVER_ADDR', 'localhost:51075') + + # TODO: create your grpc client and wait for the server to become available. + # The client may start before the server. + stub = None + + # A list of messages obtained from the server-py but not yet requested by the user to be shown + # (via the http's /getAndFlushMessages). + postbox = PostBox() + + # TODO: Implement and run a messages stream consumer in a background thread here. + # It should fetch messages via the grpc client and store them in the postbox. + + # Pass the stub and the postbox to the HTTP server. + # Dirty, but this simple http server doesn't provide interface + # for passing arguments to the handler c-tor. + MessageHandler._stub = stub + MessageHandler._postbox = postbox + + http_port = os.environ.get('MESSENGER_HTTP_PORT', '8080') + http_server_address = ('0.0.0.0', int(http_port)) + + # NB: handler_class is instantiated for every http request. Do not store any inter-request state in it. + httpd = HTTPServer(http_server_address, MessageHandler) + httpd.serve_forever() + + +if __name__ == '__main__': + main() diff --git a/homework/02-grpc-messenger/templates/messenger-py/client/requirements.txt b/homework/02-grpc-messenger/templates/messenger-py/client/requirements.txt new file mode 100644 index 0000000..192a200 --- /dev/null +++ b/homework/02-grpc-messenger/templates/messenger-py/client/requirements.txt @@ -0,0 +1,2 @@ +grpcio==1.75.0 +grpcio-tools==1.75.0 \ No newline at end of file diff --git a/homework/02-grpc-messenger/templates/messenger-py/proto/__init__.py b/homework/02-grpc-messenger/templates/messenger-py/proto/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/homework/02-grpc-messenger/templates/messenger-py/proto/messenger.proto b/homework/02-grpc-messenger/templates/messenger-py/proto/messenger.proto new file mode 100644 index 0000000..40ca766 --- /dev/null +++ b/homework/02-grpc-messenger/templates/messenger-py/proto/messenger.proto @@ -0,0 +1,5 @@ +syntax = "proto3"; + +package mes_grpc; + +// TODO: Add messages and service \ No newline at end of file diff --git a/homework/02-grpc-messenger/templates/messenger-py/server/__init__.py b/homework/02-grpc-messenger/templates/messenger-py/server/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/homework/02-grpc-messenger/tests/Dockerfile b/homework/02-grpc-messenger/tests/Dockerfile new file mode 100644 index 0000000..041f744 --- /dev/null +++ b/homework/02-grpc-messenger/tests/Dockerfile @@ -0,0 +1,21 @@ +# syntax=docker/dockerfile:1 +FROM python:3.12-alpine + +COPY tests/requirements.txt . +RUN --mount=type=cache,id=distsys-course-pip,target=/root/.cache/pip,sharing=locked \ + pip install -r requirements.txt + +ARG GRPCURL_VERSION=1.9.3 +ARG GRPCURL_SHA256=a926b62a85787ccf73ef8736b3ae554f1242e39d92bb8767a79d6dd23b11d1d5 +RUN apk add --no-cache curl tini \ + && curl --fail --silent --show-error --location \ + "https://github.com/fullstorydev/grpcurl/releases/download/v${GRPCURL_VERSION}/grpcurl_${GRPCURL_VERSION}_linux_x86_64.tar.gz" \ + --output /tmp/grpcurl.tar.gz \ + && echo "${GRPCURL_SHA256} /tmp/grpcurl.tar.gz" | sha256sum -c \ + && tar -xzf /tmp/grpcurl.tar.gz -C /usr/local/bin \ + && rm -f /tmp/grpcurl.tar.gz + +COPY solution/proto/ solution/proto/ +COPY tests/*.py tests/ + +ENTRYPOINT ["/sbin/tini", "--", "python3", "-u", "tests/main.py"] diff --git a/homework/02-grpc-messenger/tests/main.py b/homework/02-grpc-messenger/tests/main.py new file mode 100644 index 0000000..6d7dfa7 --- /dev/null +++ b/homework/02-grpc-messenger/tests/main.py @@ -0,0 +1,109 @@ +import argparse +import os +import pathlib +import signal +import subprocess +import sys + +import pytest + + +SCRIPT_DIR = pathlib.Path(__file__).parent.resolve() +SUITE_TIMEOUT_S = 120 + + +class PassedCounter: + def __init__(self): + self.passed = 0 + + def pytest_report_teststatus(self, report, config): + if report.when == 'call' and report.passed: + self.passed += 1 + + +def suite_score(passed, expected, maximum, exit_code=pytest.ExitCode.OK): + return maximum if exit_code == pytest.ExitCode.OK and passed == expected else 0 + + +def run_suite(filename, expected, maximum): + # Each component has its own deadline so a timeout cannot discard other scores. + process = subprocess.Popen( + [sys.executable, '-u', str(pathlib.Path(__file__).resolve()), + '--suite', str(SCRIPT_DIR / filename), str(expected)], + start_new_session=os.name == 'posix', + ) + try: + exit_code = process.wait(timeout=SUITE_TIMEOUT_S) + except subprocess.TimeoutExpired: + if os.name == 'posix': + # Include grpcurl and other subprocesses started by this test suite. + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass # The suite may have exited just after wait() timed out. + else: + process.kill() + process.wait() + print(f'{filename} exceeded {SUITE_TIMEOUT_S} seconds; component score is zero.', flush=True) + return 0 + return maximum if exit_code == 0 else 0 + + +def suite_exit_code(filename, expected): + counter = PassedCounter() + exit_code = pytest.main(['-vs', filename], plugins=[counter]) + return 0 if suite_score(counter.passed, expected, 1, exit_code) else 1 + + +def component_enabled(environment_name): + return os.environ.get(environment_name, '1') == '1' + + +def parse_args(): + parser = argparse.ArgumentParser() + parser.add_argument( + '--component', + choices=('all', 'proto', 'server', 'client'), + default='all', + ) + parser.add_argument('--suite', nargs=2, metavar=('FILE', 'EXPECTED'), help=argparse.SUPPRESS) + return parser.parse_args() + + +def main(): + args = parse_args() + if args.suite is not None: + return suite_exit_code(args.suite[0], int(args.suite[1])) + component = args.component + score = 0 + + if component in ('all', 'proto'): + proto_score = run_suite('test_proto.py', expected=1, maximum=2) + score += proto_score + print(f'Proto: {proto_score}/2') + print() + + if component in ('all', 'server'): + if component_enabled('MESSENGER_SERVER_TESTS_ENABLED'): + server_score = run_suite('test_server.py', expected=4, maximum=4) + else: + server_score = 0 + print('Server tests were not run because the server image did not build.') + score += server_score + print(f'Server: {server_score}/4') + print() + + if component in ('all', 'client'): + if component_enabled('MESSENGER_CLIENT_TESTS_ENABLED'): + client_score = run_suite('test_client.py', expected=3, maximum=4) + else: + client_score = 0 + print('Client tests were not run because the client image did not build.') + score += client_score + print(f'Client: {client_score}/4') + + print(f'\nSCORE: {score}') + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/homework/02-grpc-messenger/tests/requirements.txt b/homework/02-grpc-messenger/tests/requirements.txt new file mode 100644 index 0000000..4345391 --- /dev/null +++ b/homework/02-grpc-messenger/tests/requirements.txt @@ -0,0 +1,5 @@ +grpcio==1.75.0 +grpcio-tools==1.75.0 +protobuf==6.32.1 +pytest==8.4.2 +requests==2.32.5 diff --git a/homework/02-grpc-messenger/tests/test_client.py b/homework/02-grpc-messenger/tests/test_client.py new file mode 100644 index 0000000..a26ff3a --- /dev/null +++ b/homework/02-grpc-messenger/tests/test_client.py @@ -0,0 +1,179 @@ +import os +import time + +import pytest +import requests + +HTTP_CONNECT_TIMEOUT_S = 1 +HTTP_READ_TIMEOUT_S = 5 +HTTP_READY_TIMEOUT_S = 20 +HTTP_RETRY_INTERVAL_S = 0.5 +HTTP_TIMEOUT = (HTTP_CONNECT_TIMEOUT_S, HTTP_READ_TIMEOUT_S) +MESSAGE_TIMEOUT_S = 10 +MESSAGE_POLL_INTERVAL_S = 0.05 + + +def wait_for_http(url): + deadline = time.monotonic() + HTTP_READY_TIMEOUT_S + last_exception = None + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + pytest.fail( + f'timed out waiting for HTTP endpoint {url}: {last_exception}', + pytrace=False, + ) + readiness_timeout = ( + min(HTTP_CONNECT_TIMEOUT_S, remaining), + min(HTTP_READ_TIMEOUT_S, remaining), + ) + try: + response = requests.get(url, timeout=readiness_timeout) + response.close() + return + except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as exc: + last_exception = exc + + remaining = deadline - time.monotonic() + if remaining <= 0: + pytest.fail( + f'timed out waiting for HTTP endpoint {url}: {last_exception}', + pytrace=False, + ) + time.sleep(min(HTTP_RETRY_INTERVAL_S, remaining)) + + +def post_json(url, path, timeout=HTTP_TIMEOUT, **kwargs): + endpoint = url + path + try: + response = requests.post(endpoint, timeout=timeout, **kwargs) + except requests.exceptions.Timeout as exc: + pytest.fail(f'timed out waiting for HTTP response from {endpoint}: {exc}', pytrace=False) + except requests.exceptions.ConnectionError as exc: + pytest.fail(f'could not connect to HTTP endpoint {endpoint}: {exc}', pytrace=False) + + try: + assert response.status_code == 200, response.text + return response.json() + finally: + response.close() + + +@pytest.fixture(scope='session') +def client1_ready_url(): + url = 'http://' + os.environ.get('MESSENGER_TEST_CLIENT1_ADDR', '127.0.0.1:8080') + wait_for_http(url) + return url + + +@pytest.fixture(scope='session') +def client2_ready_url(): + url = 'http://' + os.environ.get('MESSENGER_TEST_CLIENT2_ADDR', '127.0.0.1:8081') + wait_for_http(url) + return url + + +@pytest.fixture +def client1_url(client1_ready_url): + url = client1_ready_url + get_messages(url) # we need to flush pending messages before and after each tests + yield url + get_messages(url) + + +@pytest.fixture +def client2_url(client2_ready_url): + url = client2_ready_url + get_messages(url) + yield url + get_messages(url) + + +def send_message(url, mes): + return post_json(url, '/sendMessage', json=mes) + + +def get_messages(url, **kwargs): + return post_json(url, '/getAndFlushMessages', **kwargs) + + +def wait_for_messages(url, expected): + deadline = time.monotonic() + MESSAGE_TIMEOUT_S + messages = [] + while len(messages) < len(expected): + remaining = deadline - time.monotonic() + if remaining <= 0: + pytest.fail( + f'timed out waiting for messages from {url}: ' + f'expected {expected!r}, received {messages!r}', + pytrace=False, + ) + # Divide the remaining budget between connecting and reading the response. + timeout = ( + min(HTTP_CONNECT_TIMEOUT_S, remaining / 2), + min(HTTP_READ_TIMEOUT_S, remaining / 2), + ) + batch = get_messages(url, timeout=timeout) + assert isinstance(batch, list), f'expected a message array, received {batch!r}' + messages.extend(batch) + assert messages == expected[:len(messages)], \ + f'expected {expected!r}, received {messages!r}' + remaining = deadline - time.monotonic() + if remaining <= 0: + pytest.fail(f'timed out waiting for messages from {url}', pytrace=False) + if len(messages) < len(expected): + time.sleep(min(MESSAGE_POLL_INTERVAL_S, remaining)) + return messages + + +def test_single_client_single_message(client1_url, client2_url): + mes = { + 'author': 'TestSingleClient', + 'text': 'This is test text' + } + resp = send_message(client1_url, mes) + mes['sendTime'] = resp['sendTime'] + assert wait_for_messages(client1_url, [mes]) == [mes] + # Drain both subscriptions before the next test sends more messages. + assert wait_for_messages(client2_url, [mes]) == [mes] + + +def test_single_client_multiple_messages(client1_url, client2_url): + mes = [{ + 'author': 'TestSingleClient1', + 'text': 'This is test text' + }, { + 'author': 'TestSingleClient2', + 'text': 'This is test text' + }] + for m in mes: + resp = send_message(client1_url, m) + m['sendTime'] = resp['sendTime'] + assert wait_for_messages(client1_url, mes) == mes + assert wait_for_messages(client2_url, mes) == mes + + +def test_two_clients_multiple_messages(client1_url, client2_url): + client1_name = 'TestMultiClient1' + client2_name = 'TestMultiClient2' + mes = [{ + 'author': client1_name, + 'text': 'This is test text #1' + }, { + 'author': client1_name, + 'text': 'This is test text #2' + }, { + 'author': client2_name, + 'text': 'This is test text #3' + }, { + 'author': client2_name, + 'text': 'This is test text #4' + }] + times = set() + for m in mes: + resp = send_message(client1_url if m['author'] == client1_name else client2_url, m) + m['sendTime'] = resp['sendTime'] + times.add(m['sendTime']) + assert len(times) == len(mes) + assert wait_for_messages(client1_url, mes) == mes + assert wait_for_messages(client2_url, mes) == mes diff --git a/homework/02-grpc-messenger/tests/test_proto.py b/homework/02-grpc-messenger/tests/test_proto.py new file mode 100644 index 0000000..5e78406 --- /dev/null +++ b/homework/02-grpc-messenger/tests/test_proto.py @@ -0,0 +1,140 @@ +import pathlib +import tempfile + +import grpc_tools +from google.protobuf import descriptor +from google.protobuf import descriptor_pb2 +from google.protobuf import descriptor_pool +from grpc_tools import protoc + + +SCRIPT_DIR = pathlib.Path(__file__).parent.resolve() +PROTO_DIR = SCRIPT_DIR.parent / 'solution' / 'proto' +PROTO_FILE = PROTO_DIR / 'messenger.proto' +WELL_KNOWN_PROTO_DIR = pathlib.Path(grpc_tools.__file__).parent / '_proto' + + +def compile_descriptor_set(output_path): + result = protoc.main([ + 'grpc_tools.protoc', + f'-I{PROTO_DIR}', + f'-I{WELL_KNOWN_PROTO_DIR}', + f'--descriptor_set_out={output_path}', + '--include_imports', + str(PROTO_FILE), + ]) + assert result == 0, 'messenger.proto must compile successfully' + + descriptor_set = descriptor_pb2.FileDescriptorSet() + descriptor_set.ParseFromString(output_path.read_bytes()) + return descriptor_set + + +def build_descriptor_pool(descriptor_set): + pool = descriptor_pool.DescriptorPool() + remaining = list(descriptor_set.file) + while remaining: + deferred = [] + for file_descriptor in remaining: + try: + pool.Add(file_descriptor) + except TypeError: + deferred.append(file_descriptor) + assert len(deferred) < len(remaining), 'messenger.proto imports could not be resolved' + remaining = deferred + return pool + + +def require_singular_field(message_type, field_name, field_type, message_type_name=None): + assert field_name in message_type.fields_by_name, \ + f'{message_type.full_name} must contain field {field_name}' + field = message_type.fields_by_name[field_name] + assert not field.is_repeated, f'{field.full_name} must be a singular field' + assert field.type == field_type, f'{field.full_name} has an invalid type' + if message_type_name is not None: + assert field.message_type is not None + assert field.message_type.full_name == message_type_name, \ + f'{field.full_name} has an invalid message type' + + +def require_fields_can_coexist(message_type, field_names): + fields_by_oneof = {} + for field_name in field_names: + field = message_type.fields_by_name[field_name] + if field.containing_oneof is None: + continue + previous_field = fields_by_oneof.setdefault(field.containing_oneof.full_name, field_name) + assert previous_field == field_name, \ + f'{message_type.full_name} fields must allow simultaneous values' + + +def test_proto_contract(): + with tempfile.TemporaryDirectory() as temporary_directory: + descriptor_path = pathlib.Path(temporary_directory) / 'messenger.pb' + descriptor_set = compile_descriptor_set(descriptor_path) + + submitted_file = next( + (file_descriptor for file_descriptor in descriptor_set.file + if pathlib.PurePosixPath(file_descriptor.name).name == PROTO_FILE.name), + None, + ) + assert submitted_file is not None, 'messenger.proto descriptor is missing' + assert submitted_file.syntax == 'proto3', 'messenger.proto must use proto3 syntax' + assert submitted_file.package == 'mes_grpc', 'messenger.proto must use package mes_grpc' + + pool = build_descriptor_pool(descriptor_set) + try: + messenger = pool.FindServiceByName('mes_grpc.MessengerServer') + except KeyError: + raise AssertionError('gRPC service must be named mes_grpc.MessengerServer') from None + + assert 'SendMessage' in messenger.methods_by_name, \ + 'MessengerServer must contain method SendMessage' + send_message = messenger.methods_by_name['SendMessage'] + assert not send_message.client_streaming and not send_message.server_streaming, \ + 'SendMessage must be unary' + require_singular_field( + send_message.input_type, + 'author', + descriptor.FieldDescriptor.TYPE_STRING, + ) + require_singular_field( + send_message.input_type, + 'text', + descriptor.FieldDescriptor.TYPE_STRING, + ) + require_fields_can_coexist(send_message.input_type, ('author', 'text')) + require_singular_field( + send_message.output_type, + 'sendTime', + descriptor.FieldDescriptor.TYPE_MESSAGE, + 'google.protobuf.Timestamp', + ) + + assert 'ReadMessages' in messenger.methods_by_name, \ + 'MessengerServer must contain method ReadMessages' + read_messages = messenger.methods_by_name['ReadMessages'] + assert not read_messages.client_streaming and read_messages.server_streaming, \ + 'ReadMessages must be a unary request with a server stream response' + assert not read_messages.input_type.fields, \ + 'ReadMessages request must not contain fields' + require_singular_field( + read_messages.output_type, + 'author', + descriptor.FieldDescriptor.TYPE_STRING, + ) + require_singular_field( + read_messages.output_type, + 'text', + descriptor.FieldDescriptor.TYPE_STRING, + ) + require_singular_field( + read_messages.output_type, + 'sendTime', + descriptor.FieldDescriptor.TYPE_MESSAGE, + 'google.protobuf.Timestamp', + ) + require_fields_can_coexist( + read_messages.output_type, + ('author', 'text', 'sendTime'), + ) diff --git a/homework/02-grpc-messenger/tests/test_server.py b/homework/02-grpc-messenger/tests/test_server.py new file mode 100644 index 0000000..a1c5735 --- /dev/null +++ b/homework/02-grpc-messenger/tests/test_server.py @@ -0,0 +1,285 @@ +import copy +import json +import os +import pathlib +import queue +import re +import socket +import subprocess +import threading +import time +from datetime import datetime, timezone +from typing import Dict + +import pytest + +test_message = {'author': 'alice', 'text': 'hello'} + +PROTO_DIR = pathlib.Path(__file__).resolve().parent.parent / 'solution' / 'proto' + +SOCKET_CONNECT_TIMEOUT_S = 1 +SERVICE_READY_TIMEOUT_S = 20 +SERVICE_RETRY_INTERVAL_S = 0.5 +GRPC_CALL_TIMEOUT_S = 5 +GRPC_PROCESS_TIMEOUT_S = 10 +GRPC_STREAM_TIMEOUT_S = 60 +PROCESS_STOP_TIMEOUT_S = 5 +MESSAGE_TIMEOUT_S = 10 + + +def wait_for_socket(host, port): + deadline = time.monotonic() + SERVICE_READY_TIMEOUT_S + last_exception = None + while True: + try: + with socket.create_connection((host, port), timeout=SOCKET_CONNECT_TIMEOUT_S): + pass + return + except OSError as exc: + last_exception = exc + + remaining = deadline - time.monotonic() + if remaining <= 0: + pytest.fail( + f'timed out waiting for TCP endpoint {host}:{port}: {last_exception}', + pytrace=False, + ) + time.sleep(min(SERVICE_RETRY_INTERVAL_S, remaining)) + + +@pytest.fixture(scope='session') +def server_addr(): + addr = os.environ.get('MESSENGER_TEST_SERVER_ADDR', '127.0.0.1:51075') + host = addr.split(':')[0] + port = int(addr.split(':')[1]) + wait_for_socket(host, port) + yield addr + + +def send_message(server_address, message: Dict[str, str]) -> Dict[str, str]: + grpcurl_cmd = ['grpcurl', + '-max-time', str(GRPC_CALL_TIMEOUT_S), + '-import-path', str(PROTO_DIR), + '-proto', 'messenger.proto', + '-d', + json.dumps(message), + '-plaintext', + server_address, + 'mes_grpc.MessengerServer/SendMessage'] + try: + completed = subprocess.run( + grpcurl_cmd, + capture_output=True, + check=False, + timeout=GRPC_PROCESS_TIMEOUT_S, + ) + except subprocess.TimeoutExpired: + pytest.fail( + f'grpcurl did not finish within {GRPC_PROCESS_TIMEOUT_S} seconds', + pytrace=False, + ) + assert completed.returncode == 0, completed.stderr + assert len(completed.stderr) == 0, completed.stderr + output_str = completed.stdout.decode('ascii') + output = json.loads(output_str) + + message_with_timestamp = copy.deepcopy(message) + message_with_timestamp['sendTime'] = output['sendTime'] + return message_with_timestamp + + +class MessageStream: + def __init__(self, server_address): + grpcurl_cmd = ['grpcurl', + '-max-time', str(GRPC_STREAM_TIMEOUT_S), + '-import-path', str(PROTO_DIR), + '-proto', 'messenger.proto', + '-plaintext', + server_address, + 'mes_grpc.MessengerServer/ReadMessages'] + self._process = subprocess.Popen( + grpcurl_cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + self._messages = queue.Queue() + self._reader_error = None + self._reader_finished = threading.Event() + self._closed = False + self._reader = threading.Thread(target=self._read_messages, daemon=True) + self._reader.start() + + def _read_messages(self): + try: + message_lines = [] + for line in self._process.stdout: + message_lines.append(line) + if line.rstrip() == '}': + self._messages.put(json.loads(''.join(message_lines))) + message_lines = [] + except Exception as exc: + self._reader_error = exc + finally: + self._reader_finished.set() + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.close() + + def _failure_detail(self): + if self._reader_error is not None: + return f'reader failed: {self._reader_error}' + returncode = self._process.poll() + if returncode is not None: + stderr = self._process.stderr.read().strip() + return f'grpcurl exited with status {returncode}: {stderr}' + return 'grpcurl is still running but produced no matching message' + + def _get_message(self, deadline, timeout_message): + while True: + if self._reader_error is not None: + raise AssertionError(self._failure_detail()) + if self._reader_finished.is_set() and self._messages.empty(): + raise AssertionError(self._failure_detail()) + + remaining = deadline - time.monotonic() + if remaining <= 0: + raise AssertionError(f'{timeout_message}: {self._failure_detail()}') + try: + return self._messages.get(timeout=min(remaining, 0.1)) + except queue.Empty: + pass + + def wait_for_message(self, expected_message, timeout, preceding_messages=None): + deadline = time.monotonic() + timeout + while True: + message = self._get_message( + deadline, + f'timed out waiting for message {expected_message}', + ) + if message == expected_message: + return + if preceding_messages is not None: + preceding_messages.append(message) + + def read_messages(self, count, timeout): + deadline = time.monotonic() + timeout + messages = [] + while len(messages) < count: + messages.append(self._get_message( + deadline, + f'timed out after receiving {len(messages)} of {count} messages', + )) + return messages + + def close(self): + if self._closed: + return + self._closed = True + if self._process.poll() is None: + self._process.terminate() + try: + self._process.wait(timeout=PROCESS_STOP_TIMEOUT_S) + except subprocess.TimeoutExpired: + self._process.kill() + try: + self._process.wait(timeout=PROCESS_STOP_TIMEOUT_S) + except subprocess.TimeoutExpired: + pytest.fail('grpcurl did not exit after SIGKILL', pytrace=False) + self._reader.join(timeout=PROCESS_STOP_TIMEOUT_S) + assert not self._reader.is_alive() + + +def wait_for_streams(server_address, streams): + preceding_messages = [[] for _ in streams] + probes = [] + for attempt in range(10): + probe = send_message( + server_address, + {'author': 'StreamProbe', 'text': f'probe #{attempt}'}, + ) + probes.append(probe) + streams_ready = True + for index, stream in enumerate(streams): + try: + stream.wait_for_message( + probe, + timeout=1, + preceding_messages=preceding_messages[index], + ) + except AssertionError: + streams_ready = False + if streams_ready: + return [ + [message for message in messages if message not in probes] + for messages in preceding_messages + ] + raise AssertionError('ReadMessages streams did not become ready') + + +def timestamp_key(timestamp): + match = re.fullmatch(r'(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d{1,9}))?Z', timestamp) + assert match is not None, f'invalid protobuf timestamp: {timestamp}' + seconds = int(datetime.strptime(match.group(1), '%Y-%m-%dT%H:%M:%S') + .replace(tzinfo=timezone.utc).timestamp()) + nanos = int((match.group(2) or '').ljust(9, '0')) + return seconds, nanos + + +def test_send_smoke(server_addr): + send_message(server_addr, test_message) + + +def test_send_returns_ascending_time(server_addr): + outputs = [] + for _ in range(10): + outputs.append(send_message(server_addr, test_message)) + + for output1, output2 in zip(outputs, outputs[1:]): + assert timestamp_key(output1['sendTime']) < timestamp_key(output2['sendTime']) + + +def test_get_messages_smoke(server_addr): + with MessageStream(server_addr) as stream: + wait_for_streams(server_addr, [stream]) + test_message_with_timestamp = send_message(server_addr, test_message) + messages = stream.read_messages(1, timeout=MESSAGE_TIMEOUT_S) + + assert len(messages) == 1 + assert messages[0] == test_message_with_timestamp + + +def test_get_only_sends_new(server_addr): + messages1 = [] + messages3 = [] + n1, n2, n3 = 2, 3, 4 + + with MessageStream(server_addr) as stream: + messages_before_ready = wait_for_streams(server_addr, [stream]) + assert messages_before_ready == [[]] + for _ in range(n1): + message = send_message(server_addr, test_message) + messages1.append(message) + messages = stream.read_messages(n1, timeout=MESSAGE_TIMEOUT_S) + + assert len(messages1) == len(messages) + for m1, m2 in zip(messages1, messages): + assert m1 == m2 + + for _ in range(n2): + send_message(server_addr, test_message) + + with MessageStream(server_addr) as stream: + messages_before_ready = wait_for_streams(server_addr, [stream]) + assert messages_before_ready == [[]] + for _ in range(n3): + message = send_message(server_addr, test_message) + messages3.append(message) + messages = stream.read_messages(n3, timeout=MESSAGE_TIMEOUT_S) + + assert len(messages3) == len(messages) + for m1, m2 in zip(messages3, messages): + assert m1 == m2