[QI2-1523] HWCS validator

.github/workflows/validation.yaml

@@ -9,7 +9,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: DavidAnson/markdownlint-cli2-action@v11
-  tests:
+  schema-check:
     name: JSON validation
     runs-on: ubuntu-latest
     steps:
@@ -19,8 +19,32 @@ jobs:
         with:
           python-version: "3.11"
       - name: Install dependencies
-        working-directory: ./tests
+        working-directory: ./tests/schema-check
         run: pip install -r requirements.txt
       - name: Run test cases
-        working-directory: ./tests
+        working-directory: ./tests/schema-check
         run: ./tests.sh
+  model-generation:
+    name: Model generation
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.11"
+      - name: Install dependencies
+        working-directory: ./tests/control-software-validator
+        run: pip install -r requirements.txt
+      - name: Run generator
+        working-directory: ./tests/control-software-validator
+        run: ./generate.sh
+      - name: Verify files are unchanged
+        run: git diff --exit-code
+      - name: MyPy
+        # Helps ensure changes in model classes are reflected in validator
+        working-directory: ./tests/control-software-validator
+        run: mypy .
+      - name: Validator dry run
+        working-directory: ./tests/control-software-validator
+        run: CSV_MODE=dry_run pytest .

.vscode/settings.json

@@ -4,5 +4,8 @@
     "editor.rulers": [
       120
     ]
-  }
-}
+  },
+  "python.analysis.extraPaths": [
+    "tests/control-software-validator"
+  ],
+}

schemas/get_dynamic/reply_failure.schema.json

@@ -18,4 +18,4 @@
   ],
   "title": "Hardware data failure",
   "type": "object"
-}
+}

tests/control-software-validator/README.md

@@ -0,0 +1,33 @@
+# Control Software Validator
+
+## Introduction
+
+This tool can be used to verify that the control software respects the Compute Runtime Schema.
+It will send correctly formatted requests to the control software and verify that responses are correctly formatted as well.
+
+## Usage
+
+After installing the dependencies in requirements.txt,
+simply run the following command in the control-software-validator directory:
+
+```bash
+pytest .
+```
+
+The tool assumes you have the control software running on the same host with its reply channel
+available on port 4203 (`tcp://localhost:4203`), and its publish channel available on port 4204 (`tcp://localhost:4204`).
+These addresses are configurable using environment variables `CSV_HWCS_REQ_ADDRESS` and `CSV_HWCS_SUB_ADDRESS`
+respectively (or directly in the test file).
+
+Setting the environment variable `CSV_MODE` to `dry_run` triggers a dry run that does not connect to the control software.
+This is useful for verifying the validator itself, as it will fail if any of the request messages are badly formatted.
+
+## Model generation
+
+The models used in the validator are automatically generated based on the jsonschemas. The generate.sh script can be run
+in the same Python environment as the tool.
+
+## Limitations
+
+There is no way in the protocol to reliably trigger a failure, so the tool can only verify the
+correct formatting of successful responses.

tests/control-software-validator/channels/request_channel.py

@@ -0,0 +1,27 @@
+from typing import Any
+
+from pydantic import BaseModel
+from typing_extensions import override
+from zmq.asyncio import Socket
+
+
+class RequestChannel:
+    def __init__(self, stream: Socket) -> None:
+        self.stream = stream
+
+    def connect(self, address: str) -> None:
+        self.stream.connect(address)
+
+    async def request(self, request: BaseModel, return_type: type[BaseModel]) -> Any:
+        # Serializes request and tests if response is correctly formatted
+        self.stream.send_string(request.model_dump_json())
+        reply = await self.stream.recv_string()
+        return return_type.model_validate_json(reply)
+
+
+class MockRequestChannel(RequestChannel):
+    @override
+    async def request(self, request: BaseModel, return_type: type[BaseModel]) -> Any:
+        # Only serializes request but does not actually send it
+        _ = request.model_dump_json()
+        return None

tests/control-software-validator/channels/subscribe_channel.py

@@ -0,0 +1,24 @@
+from typing import Any
+
+from typing_extensions import override
+from zmq.asyncio import Socket
+
+
+class SubscribeChannel:
+    def __init__(self, stream: Socket) -> None:
+        self.stream = stream
+
+    def connect(self, address: str) -> None:
+        self.stream.connect(address)
+        self.stream.subscribe("")
+
+    async def receive(self, return_type: Any) -> Any:
+        reply = await self.stream.recv_string()
+        return return_type.model_validate_json(reply)
+
+
+class MockSubscribeChannel(SubscribeChannel):
+    @override
+    async def receive(self, return_type: Any) -> Any:
+        # Does not actually receive anything
+        return None

tests/control-software-validator/generate.sh

@@ -0,0 +1,21 @@
+#!/bin/bash
+
+# Directory to search
+schemas="../../schemas"
+models="./models"
+
+# Loop through all files in the schemas directory and subdirectories
+find "$schemas" -type f -name "*.json" | while read -r file; do
+    filename=$(basename "$file" | sed 's/\..*//')
+    parentdir=$(basename "$(dirname "$file")")
+    classname="${parentdir}_${filename}"
+
+    datamodel-codegen \
+        --input "$file" \
+        --class-name $classname \
+        --disable-timestamp \
+        --input-file-type jsonschema \
+        --output "./models/$classname.py" \
+        --field-constraints \
+        --output-model-type pydantic_v2.BaseModel 
+done

tests/control-software-validator/models/README.md

@@ -0,0 +1,3 @@
+# Models
+
+This directory contains generated models based on the json schemas. Do not edit manually, use the generate.sh script instead.

tests/control-software-validator/models/execute_reply_failure.py

@@ -0,0 +1,28 @@
+# generated by datamodel-codegen:
+#   filename:  reply_failure.schema.json
+
+from __future__ import annotations
+
+from typing import Literal
+from uuid import UUID
+
+from pydantic import BaseModel, Field
+
+
+class QuantumHardwareFailureData(BaseModel):
+    error_msg: str = Field(
+        ...,
+        description='A descriptive error message to be passed on to the user.',
+        title='Error Msg',
+    )
+
+
+class ExecuteReplyFailure(BaseModel):
+    version: str = Field(..., pattern='^\\d+\\.\\d+\\.\\d$', title='Version')
+    session_id: UUID = Field(
+        ...,
+        description='An arbitrary string, filled in in the request, which is copied into the reply object.',
+        title='Session Id',
+    )
+    status: Literal['failure'] = Field(..., title='Status')
+    payload: QuantumHardwareFailureData

tests/control-software-validator/models/execute_reply_success.py

@@ -0,0 +1,36 @@
+# generated by datamodel-codegen:
+#   filename:  reply_success.schema.json
+
+from __future__ import annotations
+
+from typing import Dict, List, Literal, Optional
+from uuid import UUID
+
+from pydantic import BaseModel, Field
+
+
+class QuantumHardwareRunCircuitResult(BaseModel):
+    job_id: int = Field(
+        ..., description='Client defined identifier for the execution.', title='Job Id'
+    )
+    results: Dict[str, int] = Field(
+        ...,
+        description='Mapping of measured bitstring (little endian notation; q[n]...q[0]) to count of occurrences.',
+        title='Results',
+    )
+    raw_data: Optional[List[str]] = Field(
+        None,
+        description='A list of bitstrings (little endian notation; `q[n]...q[0]`) ordered by the shot in which it             was measured.',
+        title='Raw Data',
+    )
+
+
+class ExecuteReplySuccess(BaseModel):
+    version: str = Field(..., pattern='^\\d+\\.\\d+\\.\\d$', title='Version')
+    session_id: UUID = Field(
+        ...,
+        description='An arbitrary string, filled in in the request, which is copied into the reply object.',
+        title='Session Id',
+    )
+    status: Literal['success'] = Field(..., title='Status')
+    payload: QuantumHardwareRunCircuitResult

tests/control-software-validator/models/execute_request.py

@@ -0,0 +1,40 @@
+# generated by datamodel-codegen:
+#   filename:  request.schema.json
+
+from __future__ import annotations
+
+from typing import Literal, Optional
+from uuid import UUID
+
+from pydantic import BaseModel, Field
+
+
+class QuantumHardwareRunCircuitPayload(BaseModel):
+    job_id: int = Field(
+        ..., description='Client identifier for the execution', title='Job Id'
+    )
+    circuit: str = Field(
+        ..., description='Circuit description in cQASM language', title='Circuit'
+    )
+    number_of_shots: int = Field(
+        ...,
+        description='Number of shots to be executed for the circuit.',
+        gt=0,
+        title='Number Of Shots',
+    )
+    include_raw_data: Optional[bool] = Field(
+        False,
+        description='Whether or not to return all bitstrings in the order in which they were measured.',
+        title='Include Raw Data',
+    )
+
+
+class ExecuteRequest(BaseModel):
+    version: str = Field(..., pattern='^\\d+\\.\\d+\\.\\d$', title='Version')
+    session_id: UUID = Field(
+        ...,
+        description='An arbitrary string, filled in in the request, which is copied into the reply object.',
+        title='Session Id',
+    )
+    command: Literal['execute'] = Field(..., title='Command')
+    payload: QuantumHardwareRunCircuitPayload

tests/control-software-validator/models/get_dynamic_reply_failure.py

@@ -0,0 +1,13 @@
+# generated by datamodel-codegen:
+#   filename:  reply_failure.schema.json
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class GetDynamicReplyFailure(BaseModel):
+    version: str = Field(..., pattern='^\\d+\\.\\d+\\.\\d$', title='Version')
+    status: Literal['failure'] = Field(..., title='Status')

tests/control-software-validator/models/get_dynamic_reply_success.py

@@ -0,0 +1,20 @@
+# generated by datamodel-codegen:
+#   filename:  reply_success.schema.json
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class QuantumHardwareDynamicData(BaseModel):
+    pass
+
+
+class GetDynamicReplySuccess(BaseModel):
+    version: str = Field(..., pattern='^\\d+\\.\\d+\\.\\d$', title='Version')
+    status: Literal['success'] = Field(..., title='Status')
+    payload: QuantumHardwareDynamicData = Field(
+        ..., description='The return value(s) of the executed command.'
+    )

tests/control-software-validator/models/get_dynamic_request.py

@@ -0,0 +1,13 @@
+# generated by datamodel-codegen:
+#   filename:  request.schema.json
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class GetDynamicRequest(BaseModel):
+    version: str = Field(..., pattern='^\\d+\\.\\d+\\.\\d$', title='Version')
+    command: Literal['get_dynamic'] = Field(..., title='Command')

tests/control-software-validator/models/get_static_reply_failure.py

@@ -0,0 +1,13 @@
+# generated by datamodel-codegen:
+#   filename:  reply_failure.schema.json
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class GetStaticReplyFailure(BaseModel):
+    version: str = Field(..., pattern='^\\d+\\.\\d+\\.\\d$', title='Version')
+    status: Literal['failure'] = Field(..., title='Status')