Coding Challenge: Automating Agentic Tests and the MLOps Trinity in CI/CD

Problem Description

You are an MLOps engineer building a Continuous Integration and Continuous Deployment (CI/CD) pipeline for a dynamic AI prediction service. Standard CI/CD practices track code perfectly. Because machine learning models change based on user input your pipeline needs to accommodate Agentic Testing. You deploy AI-powered testing agents for independently analyze and execute testing workflows on live data, and

however running tests on massive live databases creates a storage edge case: you can't copy gigabytes with live user data every time the CI/CD pipeline triggers. Instead you use Query-Based Version Control.

For prevent reproducibility from breaking when the underlying database inevitably changes (e.g., a user deletes their account), your task is for build a tracking system that securely captures the MLOps Trinity (Code, Setup Data) for each pipeline run. You've got to generate a unique digital fingerprint (Data Hash) of the data fetched during the automated test so that your pipeline can verify data integrity during future debugging sessions.

Difficulty Level

Intermediate

Input & Output Specifications

Input: * run_id (string): An unique identifier for a CI/CD pipeline's testing run. * agent_code_version (string): The Git commit hash of the testing agent's logic. * setup_params (dictionary): The live environment hyperparameters (e.g., learning rates, random seeds). * query (string): The SQL database query tracked by the pipeline. * fetched_data (list of dictionaries): The exact live data rows an agent analyzed during the automated test. * current_data (list of dictionaries): Data fetched later during a pipeline verification check.

Output: * log_agent_test(): Computes the data hash for the fetched_data and stores a record containing code version setup, query, and the computed hash under the run_id. * verify_data_consistency(): Takes run_id and current_data, computes the hash about the current data. Returns True if it perfectly matches an original test's data hash, or False if the live database has drifted.

Starter Code Boilerplate

import json
import hashlib

class CICDPipelineLogger:
    def __init__(self):
        # Dictionary to store the MLOps Trinity for each run_id
        self.pipeline_logs = {}

    def _generate_data_hash(self, data_payload):
        """
        Helper method to create a unique digital fingerprint of the data.
        """
        # TODO: Implement consistent hashing for the data payload
        pass

    def log_agent_test(self, run_id, agent_code_version, setup_params, query, fetched_data):
        """
        Logs the MLOps Trinity to ensure full traceability of the Agentic Test.
        """
        # TODO: Generate a hash for fetched_data
        # TODO: Store the Code, Setup, Query, and Data Hash in self.pipeline_logs
        pass

    def verify_data_consistency(self, run_id, current_data):
        """
        Verifies if the underlying database has changed since the automated test ran.
        """
        # TODO: Retrieve the original hash from self.pipeline_logs using run_id
        # TODO: Generate a new hash for current_data
        # TODO: Return True if hashes match (data is safe), False if they differ (data drifted)
        pass

Hints

• The MLOps Trinity: Standard Git only tracks code. Your pipeline logger needs to explicitly group and store the Code (agent_code_version), Setup (setup_params), and the Data (query + the digital data hash).
• Consistent Hashing: When converting a Python list of dictionaries to a string for hashing order absolutely matters. Use json.dumps(data_payload, sort_keys=True) to ensure a payload is simply formatted exactly the same way every time before you hash it.
• Python's Hashlib: You can generate your unique digital fingerprint by encoding the consistent string and hashing it using hashlib.sha256(string_data.encode('utf-8')).hexdigest().

Test Cases

You can use the following test cases to verify your implementation:

# 1. Initialize the Pipeline Logger
logger = CICDPipelineLogger()

# Mock live data fetched by the agent during the CI/CD test
live_db_data = [
    {"user_id": 101, "action": "click", "timestamp": "2025-06-27T10:00:00"},
    {"user_id": 102, "action": "purchase", "timestamp": "2025-06-27T10:05:00"}
]

# 2. Log the Agent Test (Capturing the MLOps Trinity)
logger.log_agent_test(
    run_id="pipeline_run_8472",
    agent_code_version="git_commit_9a8b7c",
    setup_params={"learning_rate": 0.01, "random_seed": 42},
    query="SELECT * FROM user_behavior WHERE date = '2025-06-27'",
    fetched_data=live_db_data
)

# 3. Test Case A: Data is consistent (Reproducibility intact)
print("Test Case A (Consistent Data):", 
      logger.verify_data_consistency("pipeline_run_8472", live_db_data))
# Expected Output: True

# 4. Test Case B: Database changed (User 102 deleted their account)
mutated_db_data = [
    {"user_id": 101, "action": "click", "timestamp": "2025-06-27T10:00:00"}
]
print("Test Case B (Drifted Data):", 
      logger.verify_data_consistency("pipeline_run_8472", mutated_db_data))
# Expected Output: False