Add initial integration test support

.github/workflows/ci.yml

@@ -45,3 +45,32 @@ jobs:
 
     - name: Verify
       run: make verify
+
+  run-integration-check:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-python@v2
+        with:
+          python-version: '3.7'
+      - run: pip install -r requirements.txt
+        working-directory: ./integration/test_consumer/
+      - name: Build the stack
+        run: docker-compose -f docker-compose-integration.yml up -d
+        working-directory: ./integration
+      - name: Run consumer python script
+        run: python check_kafka_output.py
+        working-directory: ./integration/test_consumer
+      - name: Output filtered log (full log in artifacts)
+        run: docker-compose -f docker-compose-integration.yml logs feeds | grep -v "Sending package" | grep -v "Processing Package"
+        working-directory: ./integration/
+        if: ${{ always() }}
+      - name: Dump logs for archive
+        run: docker-compose -f docker-compose-integration.yml logs feeds > feeds-log.txt
+        working-directory: ./integration/
+        if: ${{ always() }}
+      - uses: actions/upload-artifact@v2
+        with:
+          name: package-feeds log
+          path: ./integration/feeds-log.txt
+        if: ${{ always() }}

integration/config/feeds.yml

@@ -0,0 +1,6 @@
+poll_rate: 30m
+publisher:
+  type: kafka
+  config:
+    brokers: ["kafka:9092"]
+    topic: "package-feeds"

integration/docker-compose-integration.yml

@@ -0,0 +1,37 @@
+version: "3"
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper@sha256:87314e87320abf190f0407bf1689f4827661fbb4d671a41cba62673b45b66bfa
+    ports:
+      - "2181:2181"
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+      ZOOKEEPER_SYNC_LIMIT: 2
+
+  kafka:
+    image: confluentinc/cp-kafka@sha256:c6320f9a0cbf57075e102546de110dcebdf374955f12388d58c23a54b8a47d31
+    ports:
+      - 9094:9094
+    depends_on:
+      - zookeeper
+    environment:
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_LISTENERS: INTERNAL://kafka:9092,OUTSIDE://kafka:9094
+      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,OUTSIDE://localhost:9094
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,OUTSIDE:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+      KAFKA_offsets_topic_replication_factor: 1
+
+  feeds:
+    restart: "on-failure"
+    build:
+        context: ..
+    ports:
+      - 8080:8080
+    depends_on:
+      - kafka
+    environment:
+      PACKAGE_FEEDS_CONFIG_PATH: /config/feeds.yml
+    volumes:
+      - "./config/:/config/"

integration/test_consumer/check_kafka_output.py

@@ -0,0 +1,61 @@
+from datetime import datetime
+from confluent_kafka import Consumer
+
+import time
+import requests
+
+def trigger_feeds():
+    attempts = 5
+    print("Requesting feeds poll data from registries...")
+    while True:
+        try:
+            r = requests.get('http://127.0.0.1:8080')
+            break
+        except:
+            if attempts == 0:
+                raise
+            print("Warning: Failed to request data from package-feeds, retrying after 5s")
+            time.sleep(5)
+
+        attempts -= 1
+
+    print(r.text)
+
+
+
+def main():
+    msgs = []
+
+    trigger_feeds()
+
+    c = Consumer({
+        'bootstrap.servers': '127.0.0.1:9094',
+        'group.id': 'consumer',
+        'auto.offset.reset': 'earliest'
+    })
+
+    c.subscribe(['package-feeds'])
+
+    last_poll_success = datetime.now()
+    while True:
+        msg = c.poll(2.0)
+
+        if msg is None:
+            delta = datetime.now() - last_poll_success
+            # Timeout to avoid hanging on poll() loop
+            if delta.total_seconds() > 10:
+                break
+            continue
+        if msg.error():
+            print("Consumer error: {}".format(msg.error()))
+            continue
+        msgs.append(msg)
+        last_poll_success = datetime.now()
+        print('Received message: {}'.format(msg.value().decode('utf-8')))
+
+    print(f"\n\n------------------------------------------------------------\n\nReceived a total of {len(msgs)} messages")
+    c.close()
+    assert len(msgs) > 0, "Failed to assert that atleast a single package was received"
+
+if __name__ == '__main__':
+    main()

integration/test_consumer/requirements.txt

@@ -0,0 +1,6 @@
+certifi==2021.5.30
+chardet==4.0.0
+confluent-kafka==1.7.0
+idna==2.10
+requests==2.25.1
+urllib3==1.26.5