resolve conflicts (#694)

Co-authored-by: Seher Karakuzu <[email protected]>

tiled/_tests/test_access_control.py

@@ -2,6 +2,7 @@
 
 import numpy
 import pytest
+from starlette.status import HTTP_403_FORBIDDEN
 
 from ..adapters.array import ArrayAdapter
 from ..adapters.mapping import MapAdapter
@@ -212,15 +213,15 @@ def test_writing_blocked_by_access_policy(enter_password, context):
     with enter_password("secret1"):
         alice_client = from_context(context, username="alice")
     alice_client["d"]["x"].metadata
-    with fail_with_status_code(403):
+    with fail_with_status_code(HTTP_403_FORBIDDEN):
         alice_client["d"]["x"].update_metadata(metadata={"added_key": 3})
     alice_client.logout()
 
 
 def test_create_blocked_by_access_policy(enter_password, context):
     with enter_password("secret1"):
         alice_client = from_context(context, username="alice")
-    with fail_with_status_code(403):
+    with fail_with_status_code(HTTP_403_FORBIDDEN):
         alice_client["e"].write_array([1, 2, 3])
     alice_client.logout()
 

tiled/_tests/test_allow_origins.py

@@ -1,3 +1,5 @@
+from starlette.status import HTTP_200_OK, HTTP_400_BAD_REQUEST
+
 from ..client import Context
 from ..server.app import build_app_from_config
 
@@ -25,11 +27,11 @@ def test_cors_enforcement():
     with Context.from_app(build_app_from_config(strict_config)) as context:
         request = context.http_client.build_request("OPTIONS", "/", headers=headers)
         response = context.http_client.send(request)
-        assert response.status_code == 400
+        assert response.status_code == HTTP_400_BAD_REQUEST
 
 
 def test_allow_origins():
     with Context.from_app(build_app_from_config(permissive_config)) as context:
         request = context.http_client.build_request("OPTIONS", "/", headers=headers)
         response = context.http_client.send(request)
-        assert response.status_code == 200
+        assert response.status_code == HTTP_200_OK

tiled/_tests/test_array.py

@@ -7,6 +7,7 @@
 import httpx
 import numpy
 import pytest
+from starlette.status import HTTP_400_BAD_REQUEST, HTTP_406_NOT_ACCEPTABLE
 
 from ..adapters.array import ArrayAdapter
 from ..adapters.mapping import MapAdapter
@@ -135,7 +136,7 @@ def test_block_validation(context):
     block_url = httpx.URL(client.item["links"]["block"])
     # Malformed because it has only 2 dimensions, not 3.
     malformed_block_url = block_url.copy_with(params={"block": "0,0"})
-    with fail_with_status_code(400):
+    with fail_with_status_code(HTTP_400_BAD_REQUEST):
         client.context.http_client.get(malformed_block_url).raise_for_status()
 
 
@@ -149,7 +150,7 @@ def test_dask(context):
 
 def test_array_format_shape_from_cube(context):
     client = from_context(context)["cube"]
-    with fail_with_status_code(406):
+    with fail_with_status_code(HTTP_406_NOT_ACCEPTABLE):
         hyper_cube = client["tiny_hypercube"].export("test.png")  # noqa: F841
 
 

tiled/_tests/test_asset_access.py

@@ -2,6 +2,7 @@
 from pathlib import Path
 
 import pytest
+from starlette.status import HTTP_403_FORBIDDEN
 
 from ..catalog import in_memory
 from ..client import Context, from_context
@@ -77,5 +78,5 @@ def test_do_not_expose_raw_assets(tmpdir):
     with Context.from_app(app) as context:
         client = from_context(context, include_data_sources=True)
         client.write_array([1, 2, 3], key="x")
-        with fail_with_status_code(403):
+        with fail_with_status_code(HTTP_403_FORBIDDEN):
             client["x"].raw_export(tmpdir / "exported")

tiled/_tests/test_authentication.py

@@ -7,6 +7,11 @@
 
 import numpy
 import pytest
+from starlette.status import (
+    HTTP_400_BAD_REQUEST,
+    HTTP_401_UNAUTHORIZED,
+    HTTP_422_UNPROCESSABLE_ENTITY,
+)
 
 from ..adapters.array import ArrayAdapter
 from ..adapters.mapping import MapAdapter
@@ -80,12 +85,12 @@ def test_password_auth(enter_password, config):
         client.logout()
 
         # Bob's password should not work for Alice.
-        with fail_with_status_code(401):
+        with fail_with_status_code(HTTP_401_UNAUTHORIZED):
             with enter_password("secret2"):
                 from_context(context, username="alice")
 
         # Empty password should not work.
-        with fail_with_status_code(422):
+        with fail_with_status_code(HTTP_422_UNPROCESSABLE_ENTITY):
             with enter_password(""):
                 from_context(context, username="alice")
 
@@ -302,9 +307,9 @@ def test_admin(enter_password, config):
         assert [role["name"] for role in user_roles] == ["user"]
 
         # As bob, admin functions should be disallowed.
-        with fail_with_status_code(401):
+        with fail_with_status_code(HTTP_401_UNAUTHORIZED):
             context.admin.list_principals()
-        with fail_with_status_code(401):
+        with fail_with_status_code(HTTP_401_UNAUTHORIZED):
             context.admin.show_principal(some_principal_uuid)
 
 
@@ -358,23 +363,23 @@ def test_api_key_scopes(enter_password, config):
         metrics_key_info = context.create_api_key(scopes=["metrics"])
         context.logout()
         context.api_key = metrics_key_info["secret"]
-        with fail_with_status_code(401):
+        with fail_with_status_code(HTTP_401_UNAUTHORIZED):
             from_context(context)
         context.api_key = None
 
         # Log in as ordinary user.
         with enter_password("secret2"):
             context.authenticate(username="bob")
         # Try to request a key with more scopes that the user has.
-        with fail_with_status_code(400):
+        with fail_with_status_code(HTTP_400_BAD_REQUEST):
             context.create_api_key(scopes=["admin:apikeys"])
         # Request a key with reduced scope that can *only* read metadata.
         metadata_key_info = context.create_api_key(scopes=["read:metadata"])
         context.logout()
         context.api_key = metadata_key_info["secret"]
         restricted_client = from_context(context)
         restricted_client["A1"]
-        with fail_with_status_code(401):
+        with fail_with_status_code(HTTP_401_UNAUTHORIZED):
             restricted_client["A1"].read()  # no 'read:data' scope
         context.api_key = None
 
@@ -405,7 +410,7 @@ def test_api_key_revoked(enter_password, config):
         assert len(context.whoami()["api_keys"]) == 0
         context.logout()
         context.api_key = key_info["secret"]
-        with fail_with_status_code(401):
+        with fail_with_status_code(HTTP_401_UNAUTHORIZED):
             from_context(context)
 
 
@@ -420,7 +425,7 @@ def test_api_key_expiration(enter_password, config):
         context.logout()
         context.api_key = key_info["secret"]
         time.sleep(2)
-        with fail_with_status_code(401):
+        with fail_with_status_code(HTTP_401_UNAUTHORIZED):
             from_context(context)
 
 
@@ -435,7 +440,7 @@ def test_api_key_limit(enter_password, config):
             for i in range(authentication.API_KEY_LIMIT):
                 context.create_api_key(note=f"key {i}")
             # Hit API key limit.
-            with fail_with_status_code(400):
+            with fail_with_status_code(HTTP_400_BAD_REQUEST):
                 context.create_api_key(note="one key too many")
     finally:
         authentication.API_KEY_LIMIT = original_limit
@@ -453,7 +458,7 @@ def test_session_limit(enter_password, config):
                     context.authenticate(username="alice")
                     context.logout()
                 # Hit Session limit.
-                with fail_with_status_code(400):
+                with fail_with_status_code(HTTP_400_BAD_REQUEST):
                     context.authenticate(username="alice")
     finally:
         authentication.SESSION_LIMIT = original_limit
@@ -548,7 +553,7 @@ def test_admin_api_key_any_principal(
         context.http_client.get(resource).raise_for_status()
         context.api_key = None
         # The same endpoint fails without an API key
-        with fail_with_status_code(401):
+        with fail_with_status_code(HTTP_401_UNAUTHORIZED):
             context.http_client.get(resource).raise_for_status()
 
 
@@ -583,7 +588,7 @@ def test_admin_api_key_any_principal_exceeds_scopes(enter_password, principals_c
             context.authenticate(username="alice")
 
         principal_uuid = principals_context["uuid"]["bob"]
-        with fail_with_status_code(400) as fail_info:
+        with fail_with_status_code(HTTP_400_BAD_REQUEST) as fail_info:
             context.admin.create_api_key(principal_uuid, scopes=["read:principals"])
         fail_message = " must be a subset of the principal's scopes "
         assert fail_message in fail_info.value.response.text
@@ -601,7 +606,7 @@ def test_api_key_any_principal(enter_password, principals_context, username):
             context.authenticate(username="bob")
 
         principal_uuid = principals_context["uuid"][username]
-        with fail_with_status_code(401):
+        with fail_with_status_code(HTTP_401_UNAUTHORIZED):
             context.admin.create_api_key(principal_uuid, scopes=["read:metadata"])
 
 
@@ -631,7 +636,7 @@ def test_api_key_bypass_scopes(enter_password, principals_context):
                 {"api_key": api_key, "scopes": []},
             ):
                 context.api_key = query_params.pop("api_key", None)
-                with fail_with_status_code(401):
+                with fail_with_status_code(HTTP_401_UNAUTHORIZED):
                     context.http_client.get(
                         resource, params=query_params
                     ).raise_for_status()

tiled/_tests/test_client.py

@@ -3,6 +3,7 @@
 import httpx
 import pytest
 import yaml
+from starlette.status import HTTP_400_BAD_REQUEST
 
 from ..adapters.mapping import MapAdapter
 from ..client import Context, from_context, from_profile, record_history
@@ -27,12 +28,12 @@ def test_client_version_check():
 
         # Too-old user agent should generate a 400.
         context.http_client.headers["user-agent"] = "python-tiled/0.1.0a77"
-        with fail_with_status_code(400):
+        with fail_with_status_code(HTTP_400_BAD_REQUEST):
             list(client)
 
         # Gibberish user agent should generate a 400.
         context.http_client.headers["user-agent"] = "python-tiled/gibberish"
-        with fail_with_status_code(400):
+        with fail_with_status_code(HTTP_400_BAD_REQUEST):
             list(client)
 
 

tiled/_tests/test_compression.py

@@ -1,5 +1,6 @@
 import numpy
 import pytest
+from starlette.status import HTTP_200_OK
 from starlette.testclient import TestClient
 
 from ..adapters.array import ArrayAdapter
@@ -30,8 +31,8 @@ def test_gzip_supported(app):
         data_response = client.get(
             "/api/v1/array/full/compresses_well", headers={"Accept": "text/csv"}
         )
-    assert metadata_response.status_code == 200
-    assert data_response.status_code == 200
+    assert metadata_response.status_code == HTTP_200_OK
+    assert data_response.status_code == HTTP_200_OK
     assert "gzip" in metadata_response.headers["Content-Encoding"]
     assert "gzip" in data_response.headers["Content-Encoding"]
 
@@ -44,7 +45,7 @@ def test_zstd_preferred(app):
         data_response = client.get(
             "/api/v1/array/full/compresses_well", headers={"Accept": "text/csv"}
         )
-    assert metadata_response.status_code == 200
-    assert data_response.status_code == 200
+    assert metadata_response.status_code == HTTP_200_OK
+    assert data_response.status_code == HTTP_200_OK
     assert "zstd" in metadata_response.headers["Content-Encoding"]
     assert "zstd" in data_response.headers["Content-Encoding"]

tiled/_tests/test_dataframe.py

@@ -1,6 +1,7 @@
 import numpy
 import pandas.testing
 import pytest
+from starlette.status import HTTP_400_BAD_REQUEST
 
 from ..adapters.dataframe import DataFrameAdapter
 from ..adapters.mapping import MapAdapter
@@ -200,7 +201,7 @@ def test_redundant_query_parameters(context):
         context.http_client.get(url_path, params=params).raise_for_status()
 
     # It is an error to include query parameter 'column' AND 'field'
-    with fail_with_status_code(400) as response:
+    with fail_with_status_code(HTTP_400_BAD_REQUEST) as response:
         params = original_params
         context.http_client.get(url_path, params=params).raise_for_status()
         assert "'field'" in response.text

tiled/_tests/test_directory_walker.py

@@ -8,6 +8,7 @@
 import pytest
 import tifffile
 import yaml
+from starlette.status import HTTP_415_UNSUPPORTED_MEDIA_TYPE
 
 from ..adapters.hdf5 import HDF5Adapter
 from ..adapters.tiff import TiffAdapter
@@ -323,7 +324,7 @@ def test_unknown_mimetype(tmpdir):
             is_directory=False,
             parameter="test",
         )
-        with fail_with_status_code(415):
+        with fail_with_status_code(HTTP_415_UNSUPPORTED_MEDIA_TYPE):
             client.new(
                 key="x",
                 structure_family="array",

tiled/_tests/test_metrics.py

@@ -1,6 +1,7 @@
 import re
 
 from fastapi import APIRouter
+from starlette.status import HTTP_404_NOT_FOUND, HTTP_500_INTERNAL_SERVER_ERROR
 
 from ..client import Context, from_context
 from ..server.app import build_app_from_config
@@ -43,8 +44,8 @@ def test_error_code():
         assert total_request_time(client, 500) - baseline_time[500] == 0
         client.context.http_client.raise_server_exceptions = False
         response_500 = client.context.http_client.get("/error")
-        assert response_500.status_code == 500
+        assert response_500.status_code == HTTP_500_INTERNAL_SERVER_ERROR
         assert total_request_time(client, 500) - baseline_time[500] > 0
         response_404 = client.context.http_client.get("/does_not_exist")
-        assert response_404.status_code == 404
+        assert response_404.status_code == HTTP_404_NOT_FOUND
         assert total_request_time(client, 404) - baseline_time[404] > 0

tiled/_tests/test_openapi.py

@@ -1,4 +1,5 @@
 import pytest
+from starlette.status import HTTP_200_OK
 
 from ..adapters.mapping import MapAdapter
 from ..client import Context
@@ -41,7 +42,7 @@ def test_openapi_username_password_login(context):
     invalid path, because that has happened before.
     """
     response = context.http_client.get("/openapi.json")
-    assert response.status_code == 200
+    assert response.status_code == HTTP_200_OK
     openapi = response.json()
     token_url = openapi["components"]["securitySchemes"]["OAuth2PasswordBearer"][
         "flows"

tiled/_tests/test_queries.py

@@ -7,6 +7,7 @@
 import numpy
 import pytest
 import pytest_asyncio
+from starlette.status import HTTP_400_BAD_REQUEST
 
 from ..adapters.array import ArrayAdapter
 from ..adapters.mapping import MapAdapter
@@ -165,7 +166,7 @@ def test_full_text(client):
     if client.metadata["backend"] in {"sqlite"}:
 
         def cm():
-            return fail_with_status_code(400)
+            return fail_with_status_code(HTTP_400_BAD_REQUEST)
 
     else:
         cm = nullcontext
@@ -180,7 +181,7 @@ def test_regex(client):
     if client.metadata["backend"] in {"postgresql", "sqlite"}:
 
         def cm():
-            return fail_with_status_code(400)
+            return fail_with_status_code(HTTP_400_BAD_REQUEST)
 
     else:
         cm = nullcontext

tiled/_tests/test_routes.py

@@ -1,5 +1,6 @@
 import pytest
 from httpx import AsyncClient
+from starlette.status import HTTP_200_OK
 
 from ..server.app import build_app
 
@@ -10,4 +11,4 @@ async def test_meta_routes(path):
     app = build_app({})
     async with AsyncClient(app=app, base_url="http://test") as client:
         response = await client.get(path)
-    assert response.status_code == 200
+    assert response.status_code == HTTP_200_OK