Add API to create generic results

The new /api/v3/results endpoint accepts the same data as POST
/api/v2.0/results but supports OIDC authentication and permission
control.

Other users/groups won't be able to POST results to this endpoint unless
they have a permission mapping with testcase pattern matching
"ANY-DATA:<testcase_name>" (instead of just "<testcase_name>" as in the
other v3 endpoints).

JIRA: RHELWF-11168

resultsdb/authorization.py

@@ -39,6 +39,10 @@ def match_testcase_permissions(testcase, permissions):
 
 
 def verify_authorization(user, testcase, permissions, ldap_host, ldap_searches):
+    """
+    Raises an exception if the user is not permitted to publish a result for
+    the testcase.
+    """
     if not (ldap_host and ldap_searches):
         raise InternalServerError(
             "LDAP_HOST and LDAP_SEARCHES also need to be defined if PERMISSIONS is defined"
@@ -47,7 +51,7 @@ def verify_authorization(user, testcase, permissions, ldap_host, ldap_searches):
     allowed_groups = []
     for permission in match_testcase_permissions(testcase, permissions):
         if user in permission.get("users", []):
-            return True
+            return
         allowed_groups += permission.get("groups", [])
 
     try:
@@ -67,7 +71,7 @@ def verify_authorization(user, testcase, permissions, ldap_host, ldap_searches):
     for cur_ldap_search in ldap_searches:
         groups = get_group_membership(ldap, user, con, cur_ldap_search)
         if any(g in groups for g in allowed_groups):
-            return True
+            return
         any_groups_found = any_groups_found or len(groups) > 0
 
     raise Forbidden(

resultsdb/controllers/api_v2.py

@@ -479,6 +479,14 @@ def get_result(result_id):
 @api.route("/results", methods=["POST"])
 @validate()
 def create_result(body: CreateResultParams):
+    return create_result_any_data(body)
+
+
+def create_result_any_data(body: CreateResultParams):
+    """
+    Allows creating test results with data not checked by any schema (in
+    contrast to v3 API).
+    """
     if body.data:
         invalid_keys = [key for key in body.data.keys() if ":" in key]
         if invalid_keys:

resultsdb/controllers/api_v3.py

@@ -9,9 +9,11 @@
     match_testcase_permissions,
     verify_authorization,
 )
+from resultsdb.controllers.api_v2 import create_result_any_data
 from resultsdb.controllers.common import commit_result
 from resultsdb.models import db
 from resultsdb.models.results import Result, ResultData, Testcase
+from resultsdb.parsers.api_v2 import CreateResultParams
 from resultsdb.parsers.api_v3 import (
     RESULTS_PARAMS_CLASSES,
     PermissionsParams,
@@ -26,15 +28,20 @@ def permissions():
     return app.config.get("PERMISSIONS", [])
 
 
-def _verify_authorization(user, testcase):
+def get_authorized_user(testcase) -> str:
+    """
+    Raises an exception if the current user cannot publish a result for the
+    testcase, otherwise returns the name of the current user.
+    """
+    user = app.oidc.current_token_identity[app.config["OIDC_USERNAME_FIELD"]]
     ldap_host = app.config.get("LDAP_HOST")
     ldap_searches = app.config.get("LDAP_SEARCHES")
-    return verify_authorization(user, testcase, permissions(), ldap_host, ldap_searches)
+    verify_authorization(user, testcase, permissions(), ldap_host, ldap_searches)
+    return user
 
 
 def create_result(body: ResultParamsBase):
-    user = app.oidc.current_token_identity[app.config["OIDC_USERNAME_FIELD"]]
-    _verify_authorization(user, body.testcase)
+    user = get_authorized_user(body.testcase)
 
     testcase = Testcase.query.filter_by(name=body.testcase).first()
     if not testcase:
@@ -93,10 +100,40 @@ def get_schema():
     )
 
 
+def create_any_data_endpoint(oidc, provider):
+    """
+    Creates an endpoint that accepts the same data as POST /api/v2.0/results
+    but supports OIDC authentication and permission control.
+
+    Other users/groups won't be able to POST results to this endpoint unless
+    they have a permission mapping with testcase pattern matching
+    "ANY-DATA:<testcase_name>" (instead of just "<testcase_name>" as in the
+    other v3 endpoints).
+    """
+
+    @oidc.token_auth(provider)
+    @validate()
+    # Using RootModel is a workaround for a bug in flask-pydantic that causes
+    # validation to fail with unexpected exception.
+    def create(body: RootModel[CreateResultParams]):
+        testcase = body.root.testcase["name"]
+        get_authorized_user(f"ANY-DATA:{testcase}")
+        return create_result_any_data(body.root)
+
+    api.add_url_rule(
+        "/results",
+        endpoint="results",
+        methods=["POST"],
+        view_func=create,
+    )
+
+
 def create_endpoints(oidc, provider):
     for params_class in RESULTS_PARAMS_CLASSES:
         create_endpoint(params_class, oidc, provider)
 
+    create_any_data_endpoint(oidc, provider)
+
 
 @api.route("/permissions")
 @validate()

tests/test_api_v3.py

@@ -7,6 +7,11 @@
 from resultsdb.models import db
 from resultsdb.parsers.api_v3 import RESULTS_PARAMS_CLASSES
 
+GENERIC_DATA = {
+    "testcase": {"name": "test1"},
+    "outcome": "PASSED",
+}
+
 
 @pytest.fixture(scope="function", autouse=True)
 def db_session():
@@ -498,7 +503,7 @@ def test_api_v3_bad_param_type_null(params_class, client):
 @pytest.mark.parametrize("params_class", RESULTS_PARAMS_CLASSES)
 def test_api_v3_bad_param_invalid_json(params_class, client):
     """
-    Passing unexpected JSON type must propagate an error to the user.
+    Passing invalid JSON must propagate an error to the user.
     """
     artifact_type = params_class.artifact_type()
     r = client.post(
@@ -511,7 +516,7 @@ def test_api_v3_bad_param_invalid_json(params_class, client):
 @pytest.mark.parametrize("params_class", RESULTS_PARAMS_CLASSES)
 def test_api_v3_example(params_class, client):
     """
-    Passing unexpected JSON type must propagate an error to the user.
+    All examples in schemas must be valid.
     """
     artifact_type = params_class.artifact_type()
     example = params_class.example().model_dump()
@@ -522,7 +527,7 @@ def test_api_v3_example(params_class, client):
 @pytest.mark.parametrize("params_class", RESULTS_PARAMS_CLASSES)
 def test_api_v3_missing_param(params_class, client):
     """
-    Passing unexpected JSON type must propagate an error to the user.
+    Missing a parameter must propagate an error to the user.
     """
     artifact_type = params_class.artifact_type()
     example = params_class.example().model_dump()
@@ -540,3 +545,26 @@ def test_api_v3_missing_param(params_class, client):
             }
         ]
     }
+
+
+@pytest.mark.parametrize(
+    "testcase_pattern, status_code",
+    (
+        ("ANY-DATA:*", 201),
+        ("ANY-DATA:test1", 201),
+        ("ANY-DATA:test2", 403),
+        ("ANY-DATA:", 403),
+        ("ANY-DATA", 403),
+    ),
+)
+def test_api_v3_any_data(client, permissions, testcase_pattern, status_code):
+    """
+    POST to generic endpoint with permissions to "ANY-DATA:*" testcases.
+    """
+    permission = {
+        "users": ["testuser1"],
+        "testcases": [testcase_pattern],
+    }
+    permissions.append(permission)
+    r = client.post("/api/v3/results", json=GENERIC_DATA)
+    assert r.status_code == status_code, r.text

tox.ini

@@ -3,6 +3,9 @@ envlist = py3,mypy
 requires =
     poetry
     tox-docker
+    # requests-2.32.0 breaks docker
+    # https://github.com/docker/docker-py/issues/3256
+    requests<2.32.0
 
 [pytest]
 minversion=2.0