Add domain tests and pipeline CI

.github/workflows/ci.yml

@@ -0,0 +1,36 @@
+name: CI
+
+on:
+  pull_request:
+  push:
+    branches:
+      - main
+
+permissions:
+  contents: read
+
+jobs:
+  build-and-test:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v6
+
+      - name: Install build dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install --yes libsqlite3-dev ninja-build
+
+      - name: Configure
+        run: >
+          cmake -S . -B build -G Ninja
+          -DCMAKE_BUILD_TYPE=Release
+          -DBUILD_TESTING=ON
+
+      - name: Build
+        run: cmake --build build --parallel
+
+      - name: Test
+        run: ctest --test-dir build --output-on-failure

.gitignore

@@ -1,4 +1,7 @@
 build/
+build-*/
+__pycache__/
+*.py[cod]
 data/*.db
 *.db
 *.sqlite

CMakeLists.txt

@@ -5,6 +5,7 @@ project(leo_telemetry LANGUAGES CXX)
 set(CMAKE_CXX_STANDARD 20)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 
+include(CTest)
 include(FetchContent)
 
 FetchContent_Declare(
@@ -41,4 +42,8 @@ target_link_libraries(common INTERFACE
 
 add_subdirectory(services/ingest)
 add_subdirectory(services/aggregator)
-add_subdirectory(services/controlplane)
+add_subdirectory(services/controlplane)
+
+if(BUILD_TESTING)
+  add_subdirectory(tests)
+endif()

README.md

@@ -1,5 +1,7 @@
 # TelemetryOps
 
+[![CI](https://github.com/derekk024/TelemetryOps/actions/workflows/ci.yml/badge.svg)](https://github.com/derekk024/TelemetryOps/actions/workflows/ci.yml)
+
 TelemetryOps is a small C++20 telemetry pipeline for accepting simulated satellite events, storing them in SQLite, calculating rolling health metrics, and evaluating configurable alert thresholds. It is split into three HTTP services so ingestion, aggregation, and alert evaluation can be run and inspected independently.
 
 ## Architecture
@@ -31,7 +33,7 @@ Each service also exposes process-local request counters in the Prometheus text
 - A C++20 compiler
 - SQLite 3 development headers and library
 - Git and network access during the first CMake configuration
-- Python 3 to use the optional load generator
+- Python 3 to run the integration test or optional load generator
 - `curl` to run the examples below
 
 CMake fetches these pinned dependencies during configuration:
@@ -72,6 +74,25 @@ build/services/aggregator/aggregator
 build/services/controlplane/controlplane
 ```
 
+## Test
+
+The automated suite covers both domain rules and the running service pipeline:
+
+- JSON event validation and SQLite event-ID idempotency
+- weighted packet-drop aggregation, percentile interpolation, and link-quality averaging
+- alert threshold boundaries and aggregator failures
+- real ingest and aggregator processes using a temporary SQLite database, including validation errors, duplicate suppression, satellite/window filtering, aggregate responses, and Prometheus counters
+
+Build and run every test through CTest:
+
+```bash
+cmake -S . -B build -DCMAKE_BUILD_TYPE=Debug -DBUILD_TESTING=ON
+cmake --build build --parallel
+ctest --test-dir build --output-on-failure
+```
+
+GitHub Actions runs the same build and test sequence for pull requests and pushes to `main`.
+
 ## Run locally
 
 Start the services in order. The ingest service must create the database and schema before the aggregator opens that database in read-only mode.
@@ -293,31 +314,37 @@ The final line reports responses received by the generator. It is a paced traffi
 - No retention policy or database cleanup
 - No durable control-plane configuration or alert history
 - No alert delivery mechanism beyond the HTTP response and counters
-- No automated test suite or continuous-integration workflow
 - No container or multi-host deployment configuration
 - Aggregation loads all matching window rows into memory before calculating percentiles
 - One SQLite connection per data service, without an application-level connection pool
-- `common/include/common/telemetry.hpp` is currently a placeholder; the services do not share a domain-model implementation
 
 ## Project structure
 
 ```text
 .
+├── .github/workflows/ci.yml
 ├── CMakeLists.txt
 ├── common/
-│   └── include/common/telemetry.hpp
+│   └── include/common/
+│       ├── aggregation.hpp
+│       ├── alerts.hpp
+│       └── telemetry.hpp
 ├── scripts/
 │   └── load_test.py
-└── services/
-    ├── ingest/
-    │   ├── CMakeLists.txt
-    │   └── main.cpp
-    ├── aggregator/
-    │   ├── CMakeLists.txt
-    │   └── main.cpp
-    └── controlplane/
-        ├── CMakeLists.txt
-        └── main.cpp
+├── services/
+│   ├── ingest/
+│   │   ├── CMakeLists.txt
+│   │   └── main.cpp
+│   ├── aggregator/
+│   │   ├── CMakeLists.txt
+│   │   └── main.cpp
+│   └── controlplane/
+│       ├── CMakeLists.txt
+│       └── main.cpp
+└── tests/
+    ├── CMakeLists.txt
+    ├── domain_tests.cpp
+    └── service_pipeline_test.py
 ```
 
 ## License

common/include/common/aggregation.hpp

@@ -0,0 +1,69 @@
+#pragma once
+
+#include <algorithm>
+#include <cstddef>
+#include <vector>
+
+namespace telemetry_ops {
+
+struct TelemetryRow {
+    double latency_ms;
+    int dropped;
+    int sent;
+    double link_quality;
+};
+
+struct AggregateMetrics {
+    std::size_t count = 0;
+    double drop_rate = 0.0;
+    double latency_p50_ms = 0.0;
+    double latency_p95_ms = 0.0;
+    double avg_link_quality = 0.0;
+};
+
+inline double percentile(std::vector<double> values, double percentile_value) {
+    if (values.empty()) return 0.0;
+
+    std::sort(values.begin(), values.end());
+    const double index =
+        (percentile_value / 100.0) * static_cast<double>(values.size() - 1);
+    const auto lower_index = static_cast<std::size_t>(index);
+    const double fraction = index - static_cast<double>(lower_index);
+
+    if (lower_index + 1 < values.size()) {
+        return values[lower_index] * (1.0 - fraction) +
+               values[lower_index + 1] * fraction;
+    }
+    return values[lower_index];
+}
+
+inline AggregateMetrics aggregate(const std::vector<TelemetryRow>& rows) {
+    AggregateMetrics metrics;
+    metrics.count = rows.size();
+    if (rows.empty()) return metrics;
+
+    long long sum_dropped = 0;
+    long long sum_sent = 0;
+    double sum_link_quality = 0.0;
+    std::vector<double> latencies;
+    latencies.reserve(rows.size());
+
+    for (const auto& row : rows) {
+        sum_dropped += row.dropped;
+        sum_sent += row.sent;
+        sum_link_quality += row.link_quality;
+        latencies.push_back(row.latency_ms);
+    }
+
+    if (sum_sent > 0) {
+        metrics.drop_rate =
+            static_cast<double>(sum_dropped) / static_cast<double>(sum_sent);
+    }
+    metrics.latency_p50_ms = percentile(latencies, 50.0);
+    metrics.latency_p95_ms = percentile(latencies, 95.0);
+    metrics.avg_link_quality =
+        sum_link_quality / static_cast<double>(rows.size());
+    return metrics;
+}
+
+}  // namespace telemetry_ops

common/include/common/alerts.hpp

@@ -0,0 +1,62 @@
+#pragma once
+
+#include <nlohmann/json.hpp>
+
+namespace telemetry_ops {
+
+using json = nlohmann::json;
+
+struct Thresholds {
+    double latency_p95_ms = 200.0;
+    double drop_rate = 0.05;
+    double min_link_quality = 0.7;
+    int window_s = 600;
+};
+
+inline json eval_alerts(const json& metrics, const Thresholds& thresholds) {
+    json alerts = json::array();
+
+    if (!metrics.contains("ok") || !metrics["ok"].is_boolean() ||
+        !metrics["ok"].get<bool>()) {
+        alerts.push_back({
+            {"severity", "HIGH"},
+            {"type", "AGGREGATOR_ERROR"},
+            {"message", "metrics not ok"},
+        });
+        return alerts;
+    }
+
+    if (metrics.value("count", 0) == 0) return alerts;
+
+    const double latency_p95_ms = metrics.value("latency_p95_ms", 0.0);
+    const double drop_rate = metrics.value("drop_rate", 0.0);
+    const double link_quality = metrics.value("avg_link_quality", 0.0);
+
+    if (latency_p95_ms > thresholds.latency_p95_ms) {
+        alerts.push_back({
+            {"severity", "MED"},
+            {"type", "LATENCY_P95"},
+            {"value", latency_p95_ms},
+            {"threshold", thresholds.latency_p95_ms},
+        });
+    }
+    if (drop_rate > thresholds.drop_rate) {
+        alerts.push_back({
+            {"severity", "HIGH"},
+            {"type", "DROP_RATE"},
+            {"value", drop_rate},
+            {"threshold", thresholds.drop_rate},
+        });
+    }
+    if (link_quality < thresholds.min_link_quality) {
+        alerts.push_back({
+            {"severity", "MED"},
+            {"type", "LINK_QUALITY"},
+            {"value", link_quality},
+            {"threshold", thresholds.min_link_quality},
+        });
+    }
+    return alerts;
+}
+
+}  // namespace telemetry_ops