test_artifacts.py
1 import pytest 2 from litestar.testing import TestClient 3 4 from evidently.descriptors import TextLength 5 from evidently.sdk.configs import DescriptorContent 6 7 8 @pytest.fixture 9 def created_project(test_client: TestClient): 10 """Create a test project via API.""" 11 response = test_client.post( 12 "/api/projects", 13 json={ 14 "name": "Test Project", 15 "description": "Test project for artifacts", 16 }, 17 ) 18 assert response.status_code == 201 19 project_id = response.json() 20 return project_id 21 22 23 def test_list_artifacts(test_client: TestClient, created_project): 24 """GET /api/artifacts""" 25 response = test_client.get(f"/api/artifacts?project_id={created_project}") 26 assert response.status_code == 200 27 assert response.json() == [] 28 29 # Create an artifact 30 artifact_data = {"name": "test artifact", "metadata": {}} 31 response = test_client.post( 32 f"/api/artifacts?project_id={created_project}", 33 json=artifact_data, 34 ) 35 assert response.status_code == 201 36 created_artifact = response.json() 37 38 response = test_client.get(f"/api/artifacts?project_id={created_project}") 39 assert response.status_code == 200 40 artifacts = response.json() 41 assert len(artifacts) == 1 42 assert artifacts[0]["id"] == created_artifact["id"] 43 44 45 def test_create_and_get_artifact_by_id(test_client: TestClient, created_project): 46 """GET /api/artifacts/{artifact_id}""" 47 artifact_data = {"name": "my artifact", "metadata": {}} 48 response = test_client.post( 49 f"/api/artifacts?project_id={created_project}", 50 json=artifact_data, 51 ) 52 assert response.status_code == 201 53 artifact = response.json() 54 assert artifact["name"] == "my artifact" 55 assert artifact["project_id"] == str(created_project) 56 57 response = test_client.get(f"/api/artifacts/{artifact['id']}") 58 assert response.status_code == 200 59 fetched = response.json() 60 assert fetched["id"] == artifact["id"] 61 assert fetched["name"] == "my artifact" 62 63 64 def test_get_artifact_by_name(test_client: TestClient, created_project): 65 """GET /api/artifacts/by-name/{name}""" 66 artifact_data = {"name": "some artifact", "metadata": {}} 67 response = test_client.post( 68 f"/api/artifacts?project_id={created_project}", 69 json=artifact_data, 70 ) 71 assert response.status_code == 201 72 73 response = test_client.get(f"/api/artifacts/by-name/some artifact?project_id={created_project}") 74 assert response.status_code == 200 75 fetched = response.json() 76 assert fetched["name"] == "some artifact" 77 78 response = test_client.get(f"/api/artifacts/by-name/nonexistent?project_id={created_project}") 79 assert response.status_code == 404 80 81 82 def test_list_versions(test_client: TestClient, created_project): 83 """GET /api/artifacts/{artifact_id}/versions""" 84 artifact_data = {"name": "versioned artifact", "metadata": {}} 85 response = test_client.post( 86 f"/api/artifacts?project_id={created_project}", 87 json=artifact_data, 88 ) 89 assert response.status_code == 201 90 artifact = response.json() 91 92 response = test_client.get(f"/api/artifacts/{artifact['id']}/versions") 93 assert response.status_code == 200 94 versions = response.json() 95 assert versions == [] 96 97 98 def test_create_and_get_version(test_client: TestClient, created_project): 99 """POST /api/artifacts/{artifact_id}/versions and GET /api/artifacts/{artifact_id}/versions/{version}""" 100 artifact_data = {"name": "v artifact", "metadata": {}} 101 response = test_client.post( 102 f"/api/artifacts?project_id={created_project}", 103 json=artifact_data, 104 ) 105 assert response.status_code == 201 106 artifact = response.json() 107 108 descriptor = TextLength(column_name="test_col", alias="test") 109 content = DescriptorContent.from_value(descriptor) 110 version_data = { 111 "version": 1, 112 "content": content.dict(), 113 "content_type": content.get_type().value, 114 "metadata": {}, 115 } 116 response = test_client.post( 117 f"/api/artifacts/{artifact['id']}/versions", 118 json=version_data, 119 ) 120 assert response.status_code == 201 121 response.json() 122 123 response = test_client.get(f"/api/artifacts/{artifact['id']}/versions/1") 124 assert response.status_code == 200 125 fetched = response.json() 126 assert fetched["version"] == 1 127 # Note: content comparison would need proper deserialization 128 129 130 def test_bump_artifact_version(test_client: TestClient, created_project): 131 """Test bumping artifact versions""" 132 artifact_data = {"name": "bump artifact", "metadata": {}} 133 response = test_client.post( 134 f"/api/artifacts?project_id={created_project}", 135 json=artifact_data, 136 ) 137 assert response.status_code == 201 138 artifact = response.json() 139 140 # Create first version 141 descriptor1 = TextLength(column_name="col1", alias="v1") 142 content1 = DescriptorContent.from_value(descriptor1) 143 version_data = { 144 "version": 1, 145 "content": content1.dict(), 146 "content_type": content1.get_type().value, 147 "metadata": {}, 148 } 149 response = test_client.post( 150 f"/api/artifacts/{artifact['id']}/versions", 151 json=version_data, 152 ) 153 assert response.status_code == 201 154 version1 = response.json() 155 assert version1["version"] == 1 156 157 # Create second version 158 descriptor2 = TextLength(column_name="col2", alias="v2") 159 content2 = DescriptorContent.from_value(descriptor2) 160 version_data = { 161 "version": 2, 162 "content": content2.dict(), 163 "content_type": content2.get_type().value, 164 "metadata": {}, 165 } 166 response = test_client.post( 167 f"/api/artifacts/{artifact['id']}/versions", 168 json=version_data, 169 ) 170 assert response.status_code == 201 171 version2 = response.json() 172 assert version2["version"] == 2 173 174 # Create third version 175 descriptor3 = TextLength(column_name="col3", alias="v3") 176 content3 = DescriptorContent.from_value(descriptor3) 177 version_data = { 178 "version": 3, 179 "content": content3.dict(), 180 "content_type": content3.get_type().value, 181 "metadata": {}, 182 } 183 response = test_client.post( 184 f"/api/artifacts/{artifact['id']}/versions", 185 json=version_data, 186 ) 187 assert response.status_code == 201 188 version3 = response.json() 189 assert version3["version"] == 3 190 191 192 def test_delete_artifact(test_client: TestClient, created_project): 193 """DELETE /api/artifacts/{artifact_id}""" 194 response = test_client.get(f"/api/artifacts?project_id={created_project}") 195 assert response.status_code == 200 196 assert len(response.json()) == 0 197 198 artifact_data = {"name": "test artifact", "metadata": {}} 199 response = test_client.post( 200 f"/api/artifacts?project_id={created_project}", 201 json=artifact_data, 202 ) 203 assert response.status_code == 201 204 artifact = response.json() 205 206 response = test_client.get(f"/api/artifacts?project_id={created_project}") 207 assert response.status_code == 200 208 assert len(response.json()) == 1 209 210 response = test_client.delete(f"/api/artifacts/{artifact['id']}") 211 assert response.status_code == 204 212 213 response = test_client.get(f"/api/artifacts?project_id={created_project}") 214 assert response.status_code == 200 215 assert len(response.json()) == 0 216 217 218 def test_delete_artifact_version(test_client: TestClient, created_project): 219 """DELETE /api/artifacts/artifact-versions/{artifact_version_id}""" 220 artifact_data = {"name": "test artifact", "metadata": {}} 221 response = test_client.post( 222 f"/api/artifacts?project_id={created_project}", 223 json=artifact_data, 224 ) 225 assert response.status_code == 201 226 artifact = response.json() 227 228 descriptor = TextLength(column_name="test_col", alias="test") 229 content = DescriptorContent.from_value(descriptor) 230 version_data = { 231 "version": 1, 232 "content": content.dict(), 233 "content_type": content.get_type().value, 234 "metadata": {}, 235 } 236 response = test_client.post( 237 f"/api/artifacts/{artifact['id']}/versions", 238 json=version_data, 239 ) 240 assert response.status_code == 201 241 version = response.json() 242 243 response = test_client.get(f"/api/artifacts/{artifact['id']}/versions") 244 assert response.status_code == 200 245 assert len(response.json()) == 1 246 247 response = test_client.delete(f"/api/artifacts/artifact-versions/{version['id']}") 248 assert response.status_code == 204 249 250 response = test_client.get(f"/api/artifacts/{artifact['id']}/versions") 251 assert response.status_code == 200 252 assert len(response.json()) == 0 253 254 255 def test_update_artifact(test_client: TestClient, created_project): 256 """PUT /api/artifacts/{artifact_id}""" 257 artifact_data = {"name": "test artifact", "metadata": {}} 258 response = test_client.post( 259 f"/api/artifacts?project_id={created_project}", 260 json=artifact_data, 261 ) 262 assert response.status_code == 201 263 artifact = response.json() 264 265 updated_data = { 266 "id": artifact["id"], 267 "project_id": artifact["project_id"], 268 "name": "test artifact 2", 269 "metadata": artifact["metadata"], 270 } 271 response = test_client.put( 272 f"/api/artifacts/{artifact['id']}", 273 json=updated_data, 274 ) 275 assert response.status_code == 200 276 277 response = test_client.get(f"/api/artifacts/{artifact['id']}") 278 assert response.status_code == 200 279 updated_artifact = response.json() 280 assert updated_artifact["name"] == "test artifact 2"