KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use async_finalizer method in pytest-asyncio
Best Python code snippet using pytest-asyncio_python
test_protocol.py
Source:test_protocol.py 
1"""2 Copyright 2018 Inmanta3 Licensed under the Apache License, Version 2.0 (the "License");4 you may not use this file except in compliance with the License.5 You may obtain a copy of the License at6 http://www.apache.org/licenses/LICENSE-2.07 Unless required by applicable law or agreed to in writing, software8 distributed under the License is distributed on an "AS IS" BASIS,9 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.10 See the License for the specific language governing permissions and11 limitations under the License.12 Contact: code@inmanta.com13"""14import asyncio15import base6416import datetime17import json18import os19import random20import threading21import time22import urllib.parse23import uuid24from enum import Enum25from itertools import chain26from typing import Any, Dict, Iterator, List, Optional, Union27import pydantic28import pytest29import tornado30from pydantic.types import StrictBool31from tornado import gen, web32from tornado.httpclient import AsyncHTTPClient, HTTPRequest33from tornado.httputil import url_concat34from tornado.platform.asyncio import AnyThreadEventLoopPolicy35from inmanta import config, const, protocol36from inmanta.const import ClientType37from inmanta.data.model import BaseModel38from inmanta.protocol import VersionMatch, exceptions, json_encode39from inmanta.protocol.common import (40 HTML_CONTENT,41 HTML_CONTENT_WITH_UTF8_CHARSET,42 OCTET_STREAM_CONTENT,43 ZIP_CONTENT,44 ArgOption,45 InvalidMethodDefinition,46 InvalidPathException,47 MethodProperties,48 Result,49 ReturnValue,50)51from inmanta.protocol.methods import ENV_OPTS52from inmanta.protocol.rest import CallArguments53from inmanta.protocol.return_value_meta import ReturnValueWithMeta54from inmanta.server import config as opt55from inmanta.server.config import server_bind_port56from inmanta.server.protocol import Server, ServerSlice57from inmanta.types import Apireturn58from inmanta.util import hash_file59from utils import configure60def make_random_file(size=0):61 """62 Generate a random file.63 :param size: If size is > 0 content is generated that is equal or more than size.64 """65 randomvalue = str(random.randint(0, 10000))66 if size > 0:67 while len(randomvalue) < size:68 randomvalue += randomvalue69 content = ("Hello world %s\n" % (randomvalue)).encode()70 hash = hash_file(content)71 body = base64.b64encode(content).decode("ascii")72 return (hash, content, body)73async def test_client_files(client):74 (hash, content, body) = make_random_file()75 # Check if the file exists76 result = await client.stat_file(id=hash)77 assert result.code == 40478 # Create the file79 result = await client.upload_file(id=hash, content=body)80 assert result.code == 20081 # Get the file82 result = await client.get_file(id=hash)83 assert result.code == 20084 assert "content" in result.result85 assert result.result["content"] == body86async def test_client_files_lost(client):87 (hash, content, body) = make_random_file()88 # Get the file89 result = await client.get_file(id=hash)90 assert result.code == 40491async def test_sync_client_files(client):92 # work around for https://github.com/pytest-dev/pytest-asyncio/issues/16893 asyncio.set_event_loop_policy(AnyThreadEventLoopPolicy())94 done = []95 limit = 10096 sleep = 0.0197 def do_test():98 sync_client = protocol.SyncClient("client")99 (hash, content, body) = make_random_file()100 # Check if the file exists101 result = sync_client.stat_file(id=hash)102 assert result.code == 404103 # Create the file104 result = sync_client.upload_file(id=hash, content=body)105 assert result.code == 200106 # Get the file107 result = sync_client.get_file(id=hash)108 assert result.code == 200109 assert "content" in result.result110 assert result.result["content"] == body111 done.append(True)112 thread = threading.Thread(target=do_test)113 thread.start()114 while len(done) == 0 and limit > 0:115 await gen.sleep(sleep)116 limit -= 1117 thread.join()118 assert len(done) > 0119async def test_client_files_stat(client):120 file_names = []121 i = 0122 while i < 10:123 (hash, content, body) = make_random_file()124 if hash not in file_names:125 file_names.append(hash)126 result = await client.upload_file(id=hash, content=body)127 assert result.code == 200128 i += 1129 result = await client.stat_files(files=file_names)130 assert len(result.result["files"]) == 0131 other_files = ["testtest"]132 result = await client.stat_files(files=file_names + other_files)133 assert len(result.result["files"]) == len(other_files)134async def test_diff(client):135 ca = "Hello world\n".encode()136 ha = hash_file(ca)137 result = await client.upload_file(id=ha, content=base64.b64encode(ca).decode("ascii"))138 assert result.code == 200139 cb = "Bye bye world\n".encode()140 hb = hash_file(cb)141 result = await client.upload_file(id=hb, content=base64.b64encode(cb).decode("ascii"))142 assert result.code == 200143 diff = await client.diff(ha, hb)144 assert diff.code == 200145 assert len(diff.result["diff"]) == 5146 diff = await client.diff(0, hb)147 assert diff.code == 200148 assert len(diff.result["diff"]) == 4149 diff = await client.diff(ha, 0)150 assert diff.code == 200151 assert len(diff.result["diff"]) == 4152async def test_client_files_bad(server, client):153 (hash, content, body) = make_random_file()154 # Create the file155 result = await client.upload_file(id=hash + "a", content=body)156 assert result.code == 400157async def test_client_files_corrupt(client):158 (hash, content, body) = make_random_file()159 # Create the file160 result = await client.upload_file(id=hash, content=body)161 assert result.code == 200162 state_dir = opt.state_dir.get()163 file_dir = os.path.join(state_dir, "server", "files")164 file_name = os.path.join(file_dir, hash)165 with open(file_name, "wb+") as fd:166 fd.write("Haha!".encode())167 opt.server_delete_currupt_files.set("false")168 result = await client.get_file(id=hash)169 assert result.code == 500170 result = await client.upload_file(id=hash, content=body)171 assert result.code == 500172 opt.server_delete_currupt_files.set("true")173 result = await client.get_file(id=hash)174 assert result.code == 500175 result = await client.upload_file(id=hash, content=body)176 assert result.code == 200177async def test_gzip_encoding(server):178 """179 Test if the server accepts gzipped encoding and returns gzipped encoding.180 """181 (hash, content, body) = make_random_file(size=1024)182 port = opt.get_bind_port()183 url = "http://localhost:%s/api/v1/file/%s" % (port, hash)184 zipped, body = protocol.gzipped_json({"content": body})185 assert zipped186 request = HTTPRequest(187 url=url,188 method="PUT",189 headers={"Accept-Encoding": "gzip", "Content-Encoding": "gzip"},190 body=body,191 decompress_response=True,192 )193 client = AsyncHTTPClient()194 response = await client.fetch(request)195 assert response.code == 200196 request = HTTPRequest(url=url, method="GET", headers={"Accept-Encoding": "gzip"}, decompress_response=True)197 client = AsyncHTTPClient()198 response = await client.fetch(request)199 assert response.code == 200200 assert response.headers["X-Consumed-Content-Encoding"] == "gzip"201class MainHandler(web.RequestHandler):202 def get(self):203 time.sleep(1.1)204@pytest.fixture(scope="function")205async def app(unused_tcp_port):206 http_app = web.Application([(r"/api/v1/file/abc", MainHandler)])207 server = tornado.httpserver.HTTPServer(http_app)208 server.bind(unused_tcp_port)209 server.start()210 yield server211 server.stop()212 await server.close_all_connections()213async def test_timeout_error(app):214 """215 Test test verifies that the protocol client can handle requests that timeout. This means it receives a http error216 status that is not generated by the server but by the client.217 """218 from inmanta.config import Config219 Config.load_config()220 port = str(list(app._sockets.values())[0].getsockname()[1])221 Config.set("client_rest_transport", "port", port)222 Config.set("client_rest_transport", "request_timeout", "1")223 from inmanta import protocol224 client = protocol.Client("client")225 x = await client.get_file(id="abc")226 assert x.code == 599227 assert "message" in x.result228async def test_method_properties():229 """230 Test method properties decorator and helper functions231 """232 @protocol.method(path="/test", operation="PUT", client_types=["api"], api_prefix="x", api_version=2)233 def test_method(name):234 """235 Create a new project236 """237 props = protocol.common.MethodProperties.methods["test_method"][0]238 assert "Authorization" in props.get_call_headers()239 assert props.get_listen_url() == "/x/v2/test"240 assert props.get_call_url({}) == "/x/v2/test"241async def test_invalid_client_type():242 """243 Test invalid client ype244 """245 with pytest.raises(InvalidMethodDefinition) as e:246 @protocol.method(path="/test", operation="PUT", client_types=["invalid"])247 def test_method(name):248 """249 Create a new project250 """251 assert "Invalid client type invalid specified for function" in str(e)252async def test_call_arguments_defaults():253 """254 Test processing RPC messages255 """256 @protocol.method(path="/test", operation="PUT", client_types=["api"])257 def test_method(name: str, value: int = 10):258 """259 Create a new project260 """261 call = CallArguments(protocol.common.MethodProperties.methods["test_method"][0], {"name": "test"}, {})262 await call.process()263 assert call.call_args["name"] == "test"264 assert call.call_args["value"] == 10265def test_create_client():266 with pytest.raises(AssertionError):267 protocol.SyncClient("agent", "120")268 with pytest.raises(AssertionError):269 protocol.Client("agent", "120")270async def test_pydantic():271 """272 Test validating pydantic objects273 """274 class Project(BaseModel):275 id: uuid.UUID276 name: str277 @protocol.method(path="/test", operation="PUT", client_types=["api"])278 def test_method(project: Project):279 """280 Create a new project281 """282 id = uuid.uuid4()283 call = CallArguments(284 protocol.common.MethodProperties.methods["test_method"][0], {"project": {"name": "test", "id": str(id)}}, {}285 )286 await call.process()287 project = call.call_args["project"]288 assert project.name == "test"289 assert project.id == id290 with pytest.raises(exceptions.BadRequest):291 call = CallArguments(292 protocol.common.MethodProperties.methods["test_method"][0], {"project": {"name": "test", "id": "abcd"}}, {}293 )294 await call.process()295def test_pydantic_json():296 """297 Test running pydanyic objects through the json encoder298 """299 class Options(str, Enum):300 yes = "yes"301 no = "no"302 class Project(BaseModel):303 id: uuid.UUID304 name: str305 opts: Options306 project = Project(id=uuid.uuid4(), name="test", opts="no")307 assert project.opts == Options.no308 json_string = json_encode(project)309 data = json.loads(json_string)310 assert "id" in data311 assert "name" in data312 assert data["id"] == str(project.id)313 assert data["name"] == "test"314 # Now create the project again315 new = Project(**data)316 assert project == new317 assert project is not new318async def test_pydantic_alias(unused_tcp_port, postgres_db, database_name, async_finalizer):319 """320 Round trip test on aliased object321 """322 configure(unused_tcp_port, database_name, postgres_db.port)323 class Project(BaseModel):324 source: str325 validate_: bool326 class Config:327 fields = {"validate_": {"alias": "validate"}}328 class ProjectServer(ServerSlice):329 @protocol.typedmethod(path="/test", operation="POST", client_types=["api"])330 def test_method(project: Project) -> ReturnValue[Project]: # NOQA331 """332 Create a new project333 """334 @protocol.typedmethod(path="/test2", operation="POST", client_types=["api"])335 def test_method2(project: List[Project]) -> ReturnValue[List[Project]]: # NOQA336 """337 Create a new project338 """339 @protocol.handle(test_method)340 async def test_methodi(self, project: Project) -> ReturnValue[Project]:341 new_project = project.copy()342 return ReturnValue(response=new_project)343 @protocol.handle(test_method2)344 async def test_method2i(self, project: List[Project]) -> ReturnValue[List[Project]]:345 return ReturnValue(response=project)346 rs = Server()347 server = ProjectServer(name="projectserver")348 rs.add_slice(server)349 await rs.start()350 async_finalizer.add(server.stop)351 async_finalizer.add(rs.stop)352 client = protocol.Client("client")353 projectt = Project(id=uuid.uuid4(), source="test", validate=True)354 assert projectt.validate_ is True355 projectf = Project(id=uuid.uuid4(), source="test", validate=False)356 assert projectf.validate_ is False357 async def roundtrip(obj: Project) -> None:358 data = await client.test_method(obj)359 assert obj.validate_ == data.result["data"]["validate"]360 data = await client.test_method2([obj])361 assert obj.validate_ == data.result["data"][0]["validate"]362 await roundtrip(projectf)363 await roundtrip(projectt)364async def test_return_non_warnings(unused_tcp_port, postgres_db, database_name, async_finalizer):365 """366 Test return none but pushing warnings367 """368 configure(unused_tcp_port, database_name, postgres_db.port)369 class ProjectServer(ServerSlice):370 @protocol.typedmethod(path="/test", operation="POST", client_types=["api"])371 def test_method(name: str) -> ReturnValue[None]: # NOQA372 """373 Create a new project374 """375 @protocol.handle(test_method)376 async def test_method_handler(self, name) -> ReturnValue[None]:377 rv = ReturnValue()378 rv.add_warnings(["error1", "error2"])379 return rv380 rs = Server()381 server = ProjectServer(name="projectserver")382 rs.add_slice(server)383 await rs.start()384 async_finalizer.add(server.stop)385 async_finalizer.add(rs.stop)386 client = protocol.Client("client")387 response = await client.test_method("x")388 assert response.code == 200389 assert "data" in response.result390 assert response.result["data"] is None391 assert "metadata" in response.result392 assert "warnings" in response.result["metadata"]393 assert "error1" in response.result["metadata"]["warnings"]394async def test_invalid_handler():395 """396 Handlers should be async397 """398 with pytest.raises(ValueError):399 class ProjectServer(ServerSlice):400 @protocol.method(path="/test", operation="POST", client_types=["api"])401 def test_method(self):402 """403 Create a new project404 """405 @protocol.handle(test_method)406 def test_method(self):407 return408async def test_return_value(unused_tcp_port, postgres_db, database_name, async_finalizer):409 """410 Test the use and validation of methods that use common.ReturnValue411 """412 configure(unused_tcp_port, database_name, postgres_db.port)413 class Project(BaseModel):414 id: uuid.UUID415 name: str416 class ProjectServer(ServerSlice):417 @protocol.method(path="/test", operation="POST", client_types=["api"])418 def test_method(project: Project) -> ReturnValue[Project]: # NOQA419 """420 Create a new project421 """422 @protocol.handle(test_method)423 async def test_method(self, project: Project) -> ReturnValue[Project]:424 new_project = project.copy()425 return ReturnValue(response=new_project)426 rs = Server()427 server = ProjectServer(name="projectserver")428 rs.add_slice(server)429 await rs.start()430 async_finalizer.add(server.stop)431 async_finalizer.add(rs.stop)432 client = protocol.Client("client")433 result = await client.test_method({"name": "test", "id": str(uuid.uuid4())})434 assert result.code == 200435 assert "id" in result.result436 assert "name" in result.result437async def test_return_model(unused_tcp_port, postgres_db, database_name, async_finalizer):438 """439 Test the use and validation of methods that use common.ReturnValue440 """441 configure(unused_tcp_port, database_name, postgres_db.port)442 class Project(BaseModel):443 id: uuid.UUID444 name: str445 class ProjectServer(ServerSlice):446 @protocol.method(path="/test", operation="POST", client_types=["api"])447 def test_method(project: Project) -> Project: # NOQA448 """449 Create a new project450 """451 @protocol.method(path="/test2", operation="POST", client_types=["api"])452 def test_method2(project: Project) -> None: # NOQA453 pass454 @protocol.method(path="/test3", operation="POST", client_types=["api"])455 def test_method3(project: Project) -> None: # NOQA456 pass457 @protocol.handle(test_method)458 async def test_method(self, project: Project) -> Project:459 new_project = project.copy()460 return new_project461 @protocol.handle(test_method2)462 async def test_method2(self, project: Project) -> None:463 pass464 @protocol.handle(test_method3)465 async def test_method3(self, project: Project) -> None:466 return 1467 rs = Server()468 server = ProjectServer(name="projectserver")469 rs.add_slice(server)470 await rs.start()471 async_finalizer.add(server.stop)472 async_finalizer.add(rs.stop)473 client = protocol.Client("client")474 result = await client.test_method({"name": "test", "id": str(uuid.uuid4())})475 assert result.code == 200476 assert "id" in result.result477 assert "name" in result.result478 result = await client.test_method2({"name": "test", "id": str(uuid.uuid4())})479 assert result.code == 200480 result = await client.test_method3({"name": "test", "id": str(uuid.uuid4())})481 assert result.code == 500482async def test_data_envelope(unused_tcp_port, postgres_db, database_name, async_finalizer):483 """484 Test the use and validation of methods that use common.ReturnValue485 """486 configure(unused_tcp_port, database_name, postgres_db.port)487 class Project(BaseModel):488 id: uuid.UUID489 name: str490 class ProjectServer(ServerSlice):491 @protocol.typedmethod(path="/test", operation="POST", client_types=["api"])492 def test_method(project: Project) -> ReturnValue[Project]: # NOQA493 pass494 @protocol.handle(test_method)495 async def test_method(self, project: Project) -> ReturnValue[Project]:496 new_project = project.copy()497 return ReturnValue(response=new_project)498 @protocol.typedmethod(path="/test2", operation="POST", client_types=["api"], envelope_key="method")499 def test_method2(project: Project) -> ReturnValue[Project]: # NOQA500 pass501 @protocol.handle(test_method2)502 async def test_method2(self, project: Project) -> ReturnValue[Project]:503 new_project = project.copy()504 return ReturnValue(response=new_project)505 @protocol.method(path="/test3", operation="POST", client_types=["api"], envelope=True)506 def test_method3(project: Project): # NOQA507 pass508 @protocol.handle(test_method3)509 async def test_method3(self, project: dict) -> Apireturn:510 return 200, {"id": 1, "name": 2}511 @protocol.method(path="/test4", operation="POST", client_types=["api"], envelope=True, envelope_key="project")512 def test_method4(project: Project): # NOQA513 pass514 @protocol.handle(test_method4)515 async def test_method4(self, project: dict) -> Apireturn:516 return 200, {"id": 1, "name": 2}517 rs = Server()518 server = ProjectServer(name="projectserver")519 rs.add_slice(server)520 await rs.start()521 async_finalizer.add(server.stop)522 async_finalizer.add(rs.stop)523 client = protocol.Client("client")524 # 1525 result = await client.test_method({"name": "test", "id": str(uuid.uuid4())})526 assert result.code == 200527 assert "data" in result.result528 assert "id" in result.result["data"]529 assert "name" in result.result["data"]530 # 2531 result = await client.test_method2({"name": "test", "id": str(uuid.uuid4())})532 assert result.code == 200533 assert "method" in result.result534 assert "id" in result.result["method"]535 assert "name" in result.result["method"]536 # 3537 result = await client.test_method3({"name": "test", "id": str(uuid.uuid4())})538 assert result.code == 200539 assert "data" in result.result540 assert "id" in result.result["data"]541 assert "name" in result.result["data"]542 # 4543 result = await client.test_method4({"name": "test", "id": str(uuid.uuid4())})544 assert result.code == 200545 assert "project" in result.result546 assert "id" in result.result["project"]547 assert "name" in result.result["project"]548async def test_invalid_paths():549 """550 Test path validation551 """552 with pytest.raises(InvalidPathException) as e:553 @protocol.method(path="test", operation="PUT", client_types=["api"], api_prefix="x", api_version=2)554 def test_method(name):555 pass556 assert "test should start with a /" == str(e.value)557 with pytest.raises(InvalidPathException) as e:558 @protocol.method(path="/test/<othername>", operation="PUT", client_types=["api"], api_prefix="x", api_version=2)559 def test_method2(name):560 pass561 assert str(e.value).startswith("Variable othername in path /test/<othername> is not defined in function")562async def test_nested_paths(unused_tcp_port, postgres_db, database_name, async_finalizer):563 """Test overlapping path definition"""564 configure(unused_tcp_port, database_name, postgres_db.port)565 class Project(BaseModel):566 name: str567 class ProjectServer(ServerSlice):568 @protocol.typedmethod(path="/test/<data>", operation="GET", client_types=["api"])569 def test_method(data: str) -> Project: # NOQA570 pass571 @protocol.typedmethod(path="/test/<data>/config", operation="GET", client_types=["api"])572 def test_method2(data: str) -> Project: # NOQA573 pass574 @protocol.handle(test_method)575 async def test_method(self, data: str) -> Project:576 # verify that URL encoded data is properly decoded577 assert "%20" not in data578 return Project(name="test_method")579 @protocol.handle(test_method2)580 async def test_method2(self, data: str) -> Project:581 return Project(name="test_method2")582 rs = Server()583 server = ProjectServer(name="projectserver")584 rs.add_slice(server)585 await rs.start()586 async_finalizer.add(server.stop)587 async_finalizer.add(rs.stop)588 client = protocol.Client("client")589 result = await client.test_method({"data": "test "})590 assert result.code == 200591 assert "test_method" == result.result["data"]["name"]592 client = protocol.Client("client")593 result = await client.test_method2({"data": "test"})594 assert result.code == 200595 assert "test_method2" == result.result["data"]["name"]596async def test_list_basemodel_argument(unused_tcp_port, postgres_db, database_name, async_finalizer):597 """Test list of basemodel arguments and primitive types"""598 configure(unused_tcp_port, database_name, postgres_db.port)599 class Project(BaseModel):600 name: str601 class ProjectServer(ServerSlice):602 @protocol.typedmethod(path="/test", operation="POST", client_types=["api"])603 def test_method(data: List[Project], data2: List[int]) -> Project: # NOQA604 pass605 @protocol.handle(test_method)606 async def test_method(self, data: List[Project], data2: List[int]) -> Project:607 assert len(data) == 1608 assert data[0].name == "test"609 assert len(data2) == 3610 return Project(name="test_method")611 rs = Server()612 server = ProjectServer(name="projectserver")613 rs.add_slice(server)614 await rs.start()615 async_finalizer.add(server.stop)616 async_finalizer.add(rs.stop)617 client = protocol.Client("client")618 result = await client.test_method(data=[{"name": "test"}], data2=[1, 2, 3])619 assert result.code == 200620 assert "test_method" == result.result["data"]["name"]621async def test_dict_basemodel_argument(unused_tcp_port, postgres_db, database_name, async_finalizer):622 """Test dict of basemodel arguments and primitive types"""623 configure(unused_tcp_port, database_name, postgres_db.port)624 class Project(BaseModel):625 name: str626 class ProjectServer(ServerSlice):627 @protocol.typedmethod(path="/test", operation="POST", client_types=["api"])628 def test_method(data: Dict[str, Project], data2: Dict[str, int]) -> Project: # NOQA629 pass630 @protocol.handle(test_method)631 async def test_method(self, data: Dict[str, Project], data2: Dict[str, int]) -> Project:632 assert len(data) == 1633 assert data["projectA"].name == "test"634 assert len(data2) == 3635 return Project(name="test_method")636 rs = Server()637 server = ProjectServer(name="projectserver")638 rs.add_slice(server)639 await rs.start()640 async_finalizer.add(server.stop)641 async_finalizer.add(rs.stop)642 client = protocol.Client("client")643 result = await client.test_method(data={"projectA": {"name": "test"}}, data2={"1": 1, "2": 2, "3": 3})644 assert result.code == 200645 assert "test_method" == result.result["data"]["name"]646async def test_dict_with_optional_values(unused_tcp_port, postgres_db, database_name, async_finalizer):647 """Test dict which may have None as a value"""648 configure(unused_tcp_port, database_name, postgres_db.port)649 types = Union[pydantic.StrictInt, pydantic.StrictStr]650 class Result(BaseModel):651 val: Optional[types]652 class ProjectServer(ServerSlice):653 @protocol.typedmethod(path="/test", operation="POST", client_types=["api"])654 def test_method(data: Dict[str, Optional[types]]) -> Result: # NOQA655 pass656 @protocol.handle(test_method)657 async def test_method(self, data: Dict[str, Optional[types]]) -> Result:658 assert len(data) == 1659 assert "test" in data660 return Result(val=data["test"])661 @protocol.typedmethod(path="/test", operation="GET", client_types=["api"])662 def test_method2(data: Optional[str] = None) -> None: # NOQA663 pass664 @protocol.handle(test_method2)665 async def test_method2(self, data: Optional[str] = None) -> None:666 assert data is None667 rs = Server()668 server = ProjectServer(name="projectserver")669 rs.add_slice(server)670 await rs.start()671 async_finalizer.add(server.stop)672 async_finalizer.add(rs.stop)673 client = protocol.Client("client")674 result = await client.test_method(data={"test": None})675 assert result.code == 200676 assert result.result["data"]["val"] is None677 result = await client.test_method(data={"test": 5})678 assert result.code == 200679 assert result.result["data"]["val"] == 5680 result = await client.test_method(data={"test": "test123"})681 assert result.code == 200682 assert result.result["data"]["val"] == "test123"683 result = await client.test_method2()684 assert result.code == 200685 result = await client.test_method2(data=None)686 assert result.code == 200687async def test_dict_and_list_return(unused_tcp_port, postgres_db, database_name, async_finalizer):688 """Test list of basemodel arguments"""689 configure(unused_tcp_port, database_name, postgres_db.port)690 class Project(BaseModel):691 name: str692 class ProjectServer(ServerSlice):693 @protocol.typedmethod(path="/test", operation="POST", client_types=["api"])694 def test_method(data: Project) -> List[Project]: # NOQA695 pass696 @protocol.handle(test_method)697 async def test_method(self, data: Project) -> List[Project]: # NOQA698 return [Project(name="test_method")]699 @protocol.typedmethod(path="/test2", operation="POST", client_types=["api"])700 def test_method2(data: Project) -> List[str]: # NOQA701 pass702 @protocol.handle(test_method2)703 async def test_method2(self, data: Project) -> List[str]: # NOQA704 return ["test_method"]705 rs = Server()706 server = ProjectServer(name="projectserver")707 rs.add_slice(server)708 await rs.start()709 async_finalizer.add(server.stop)710 async_finalizer.add(rs.stop)711 client = protocol.Client("client")712 result = await client.test_method(data={"name": "test"})713 assert result.code == 200714 assert len(result.result["data"]) == 1715 assert "test_method" == result.result["data"][0]["name"]716 result = await client.test_method2(data={"name": "test"})717 assert result.code == 200718 assert len(result.result["data"]) == 1719 assert "test_method" == result.result["data"][0]720async def test_method_definition():721 """722 Test typed methods with wrong annotations723 """724 with pytest.raises(InvalidMethodDefinition) as e:725 @protocol.typedmethod(path="/test", operation="PUT", client_types=["api"])726 def test_method1(name) -> None:727 """728 Create a new project729 """730 assert "has no type annotation." in str(e.value)731 with pytest.raises(InvalidMethodDefinition) as e:732 @protocol.typedmethod(path="/test", operation="PUT", client_types=["api"])733 def test_method2(name: Iterator[str]) -> None:734 """735 Create a new project736 """737 assert "Type typing.Iterator[str] of argument name can only be generic List or Dict" in str(e.value)738 with pytest.raises(InvalidMethodDefinition) as e:739 @protocol.typedmethod(path="/test", operation="PUT", client_types=["api"])740 def test_method3(name: List[object]) -> None:741 """742 Create a new project743 """744 assert (745 "Type object of argument name must be a either BaseModel, Enum, UUID, str, float, int, StrictNonIntBool, datetime, "746 "bytes or a List of these types or a Dict with str keys and values of these types."747 ) in str(e.value)748 with pytest.raises(InvalidMethodDefinition) as e:749 @protocol.typedmethod(path="/test", operation="PUT", client_types=["api"])750 def test_method4(name: Dict[int, str]) -> None:751 """752 Create a new project753 """754 assert "Type typing.Dict[int, str] of argument name must be a Dict with str keys and not int" in str(e.value)755 with pytest.raises(InvalidMethodDefinition) as e:756 @protocol.typedmethod(path="/test", operation="PUT", client_types=["api"])757 def test_method5(name: Dict[str, object]) -> None:758 """759 Create a new project760 """761 assert (762 "Type object of argument name must be a either BaseModel, Enum, UUID, str, float, int, StrictNonIntBool, datetime, "763 "bytes or a List of these types or a Dict with str keys and values of these types."764 ) in str(e.value)765 @protocol.typedmethod(path="/service_types/<service_type>", operation="DELETE", client_types=["api"])766 def lcm_service_type_delete(tid: uuid.UUID, service_type: str) -> None:767 """Delete an existing service type."""768def test_optional():769 @protocol.typedmethod(path="/service_types/<service_type>", operation="DELETE", client_types=["api"])770 def lcm_service_type_delete(tid: uuid.UUID, service_type: str, version: Optional[str] = None) -> None:771 """Delete an existing service type."""772async def test_union_types(unused_tcp_port, postgres_db, database_name, async_finalizer):773 """Test use of union types"""774 configure(unused_tcp_port, database_name, postgres_db.port)775 SimpleTypes = Union[float, int, StrictBool, str] # NOQA776 AttributeTypes = Union[SimpleTypes, List[SimpleTypes], Dict[str, SimpleTypes]] # NOQA777 class ProjectServer(ServerSlice):778 @protocol.typedmethod(path="/test", operation="GET", client_types=["api"])779 def test_method(data: SimpleTypes, version: Optional[int] = None) -> List[SimpleTypes]: # NOQA780 pass781 @protocol.handle(test_method)782 async def test_method(self, data: SimpleTypes, version: Optional[int] = None) -> List[SimpleTypes]: # NOQA783 if isinstance(data, list):784 return data785 return [data]786 @protocol.typedmethod(path="/testp", operation="POST", client_types=["api"])787 def test_methodp(data: AttributeTypes, version: Optional[int] = None) -> List[SimpleTypes]: # NOQA788 pass789 @protocol.handle(test_methodp)790 async def test_methodp(self, data: AttributeTypes, version: Optional[int] = None) -> List[SimpleTypes]: # NOQA791 if isinstance(data, list):792 return data793 return [data]794 rs = Server()795 server = ProjectServer(name="projectserver")796 rs.add_slice(server)797 await rs.start()798 async_finalizer.add(server.stop)799 async_finalizer.add(rs.stop)800 client = protocol.Client("client")801 result = await client.test_methodp(data=[5], version=7)802 assert result.code == 200803 assert len(result.result["data"]) == 1804 assert 5 == result.result["data"][0]805 result = await client.test_method(data=5, version=3)806 assert result.code == 200807 assert len(result.result["data"]) == 1808 assert 5 == result.result["data"][0]809 result = await client.test_method(data=5)810 assert result.code == 200811 assert len(result.result["data"]) == 1812 assert 5 == result.result["data"][0]813 result = await client.test_method(data=5, version=7)814 assert result.code == 200815 assert len(result.result["data"]) == 1816 assert 5 == result.result["data"][0]817async def test_basemodel_validation(unused_tcp_port, postgres_db, database_name, async_finalizer):818 """Test validation of basemodel arguments and return, and how they are reported"""819 configure(unused_tcp_port, database_name, postgres_db.port)820 class Project(BaseModel):821 name: str822 value: str823 class ProjectServer(ServerSlice):824 @protocol.typedmethod(path="/test", operation="POST", client_types=["api"])825 def test_method(data: Project) -> Project: # NOQA826 pass827 @protocol.handle(test_method)828 async def test_method(self, data: Project) -> Project: # NOQA829 return Project()830 rs = Server()831 server = ProjectServer(name="projectserver")832 rs.add_slice(server)833 await rs.start()834 async_finalizer.add(server.stop)835 async_finalizer.add(rs.stop)836 client = protocol.Client("client")837 # Check validation of arguments838 result = await client.test_method(data={})839 assert result.code == 400840 assert "error_details" in result.result841 details = result.result["error_details"]["validation_errors"]842 assert len(details) == 2843 name = [d for d in details if d["loc"] == ["data", "name"]][0]844 value = [d for d in details if d["loc"] == ["data", "value"]][0]845 assert name["msg"] == "field required"846 assert value["msg"] == "field required"847 # Check the validation of the return value848 result = await client.test_method(data={"name": "X", "value": "Y"})849 assert result.code == 500850 assert "data validation error" in result.result["message"]851async def test_ACOA_header(server):852 """853 Test if the server accepts gzipped encoding and returns gzipped encoding.854 """855 port = opt.get_bind_port()856 url = f"http://localhost:{port}/api/v1/environment"857 request = HTTPRequest(url=url, method="GET")858 client = AsyncHTTPClient()859 response = await client.fetch(request)860 assert response.code == 200861 assert response.headers.get("Access-Control-Allow-Origin") is None862 config.Config.set("server", "access-control-allow-origin", "*")863 response = await client.fetch(request)864 assert response.code == 200865 assert response.headers.get("Access-Control-Allow-Origin") == "*"866async def test_multi_version_method(unused_tcp_port, postgres_db, database_name, async_finalizer):867 """Test multi version methods"""868 configure(unused_tcp_port, database_name, postgres_db.port)869 class Project(BaseModel):870 name: str871 value: str872 class ProjectServer(ServerSlice):873 @protocol.typedmethod(path="/test2", operation="POST", client_types=["api"], api_version=3)874 @protocol.typedmethod(path="/test", operation="POST", client_types=["api"], api_version=2, envelope_key="data")875 @protocol.typedmethod(path="/test", operation="POST", client_types=["api"], api_version=1, envelope_key="project")876 def test_method(project: Project) -> Project: # NOQA877 pass878 @protocol.handle(test_method)879 async def test_method(self, project: Project) -> Project: # NOQA880 return project881 rs = Server()882 server = ProjectServer(name="projectserver")883 rs.add_slice(server)884 await rs.start()885 async_finalizer.add(server.stop)886 async_finalizer.add(rs.stop)887 # rest call888 port = opt.get_bind_port()889 request = HTTPRequest(890 url=f"http://localhost:{port}/api/v1/test", method="POST", body=json_encode({"project": {"name": "a", "value": "b"}})891 )892 client = AsyncHTTPClient()893 response = await client.fetch(request)894 assert response.code == 200895 body = json.loads(response.body)896 assert "project" in body897 request = HTTPRequest(898 url=f"http://localhost:{port}/api/v2/test", method="POST", body=json_encode({"project": {"name": "a", "value": "b"}})899 )900 client = AsyncHTTPClient()901 response = await client.fetch(request)902 assert response.code == 200903 body = json.loads(response.body)904 assert "data" in body905 request = HTTPRequest(906 url=f"http://localhost:{port}/api/v3/test2", method="POST", body=json_encode({"project": {"name": "a", "value": "b"}})907 )908 client = AsyncHTTPClient()909 response = await client.fetch(request)910 assert response.code == 200911 body = json.loads(response.body)912 assert "data" in body913 # client based calls914 client = protocol.Client("client")915 response = await client.test_method(project=Project(name="a", value="b"))916 assert response.code == 200917 assert "project" in response.result918 client = protocol.Client("client", version_match=VersionMatch.highest)919 response = await client.test_method(project=Project(name="a", value="b"))920 assert response.code == 200921 assert "data" in response.result922 client = protocol.Client("client", version_match=VersionMatch.exact, exact_version=1)923 response = await client.test_method(project=Project(name="a", value="b"))924 assert response.code == 200925 assert "project" in response.result926 client = protocol.Client("client", version_match=VersionMatch.exact, exact_version=2)927 response = await client.test_method(project=Project(name="a", value="b"))928 assert response.code == 200929 assert "data" in response.result930async def test_multi_version_handler(unused_tcp_port, postgres_db, database_name, async_finalizer):931 """Test multi version methods"""932 configure(unused_tcp_port, database_name, postgres_db.port)933 class Project(BaseModel):934 name: str935 value: str936 class ProjectServer(ServerSlice):937 @protocol.typedmethod(path="/test", operation="POST", client_types=["api"], api_version=2, envelope_key="data")938 @protocol.typedmethod(path="/test", operation="POST", client_types=["api"], api_version=1, envelope_key="project")939 def test_method(project: Project) -> Project: # NOQA940 pass941 @protocol.handle(test_method, api_version=1)942 async def test_methodX(self, project: Project) -> Project: # NOQA943 return Project(name="v1", value="1")944 @protocol.handle(test_method, api_version=2)945 async def test_methodY(self, project: Project) -> Project: # NOQA946 return Project(name="v2", value="2")947 rs = Server()948 server = ProjectServer(name="projectserver")949 rs.add_slice(server)950 await rs.start()951 async_finalizer.add(server.stop)952 async_finalizer.add(rs.stop)953 # client based calls954 client = protocol.Client("client")955 response = await client.test_method(project=Project(name="a", value="b"))956 assert response.code == 200957 assert "project" in response.result958 assert response.result["project"]["name"] == "v1"959 client = protocol.Client("client", version_match=VersionMatch.highest)960 response = await client.test_method(project=Project(name="a", value="b"))961 assert response.code == 200962 assert "data" in response.result963 assert response.result["data"]["name"] == "v2"964async def test_simple_return_type(unused_tcp_port, postgres_db, database_name, async_finalizer):965 """Test methods with simple return types"""966 configure(unused_tcp_port, database_name, postgres_db.port)967 class ProjectServer(ServerSlice):968 @protocol.typedmethod(path="/test", operation="POST", client_types=["api"])969 def test_method(project: str) -> str: # NOQA970 pass971 @protocol.handle(test_method)972 async def test_methodY(self, project: str) -> str: # NOQA973 return project974 rs = Server()975 server = ProjectServer(name="projectserver")976 rs.add_slice(server)977 await rs.start()978 async_finalizer.add(server.stop)979 async_finalizer.add(rs.stop)980 # client based calls981 client = protocol.Client("client")982 response = await client.test_method(project="x")983 assert response.code == 200984 assert response.result["data"] == "x"985async def test_html_content_type(unused_tcp_port, postgres_db, database_name, async_finalizer):986 """Test whether API endpoints with a text/html content-type work."""987 configure(unused_tcp_port, database_name, postgres_db.port)988 html_content = "<html><body>test</body></html>"989 @protocol.typedmethod(path="/test", operation="GET", client_types=["api"])990 def test_method() -> ReturnValue[str]: # NOQA991 pass992 class TestServer(ServerSlice):993 @protocol.handle(test_method)994 async def test_methodY(self) -> ReturnValue[str]: # NOQA995 return ReturnValue(response=html_content, content_type=HTML_CONTENT)996 rs = Server()997 server = TestServer(name="testserver")998 rs.add_slice(server)999 await rs.start()1000 async_finalizer.add(server.stop)1001 async_finalizer.add(rs.stop)1002 # client based calls1003 client = protocol.Client("client")1004 response = await client.test_method()1005 assert response.code == 2001006 assert response.result == html_content1007async def test_html_content_type_with_utf8_encoding(unused_tcp_port, postgres_db, database_name, async_finalizer):1008 """Test whether API endpoints with a "text/html; charset=UTF-8" content-type work."""1009 configure(unused_tcp_port, database_name, postgres_db.port)1010 html_content = "<html><body>test</body></html>"1011 @protocol.typedmethod(path="/test", operation="GET", client_types=["api"])1012 def test_method() -> ReturnValue[str]: # NOQA1013 pass1014 class TestServer(ServerSlice):1015 @protocol.handle(test_method)1016 async def test_methodY(self) -> ReturnValue[str]: # NOQA1017 return ReturnValue(response=html_content, content_type=HTML_CONTENT_WITH_UTF8_CHARSET)1018 rs = Server()1019 server = TestServer(name="testserver")1020 rs.add_slice(server)1021 await rs.start()1022 async_finalizer.add(server.stop)1023 async_finalizer.add(rs.stop)1024 # client based calls1025 client = protocol.Client("client")1026 response = await client.test_method()1027 assert response.code == 2001028 assert response.result == html_content1029async def test_octet_stream_content_type(unused_tcp_port, postgres_db, database_name, async_finalizer):1030 """Test whether API endpoints with an application/octet-stream content-type work."""1031 configure(unused_tcp_port, database_name, postgres_db.port)1032 byte_stream = b"test123"1033 @protocol.typedmethod(path="/test", operation="GET", client_types=["api"])1034 def test_method() -> ReturnValue[bytes]: # NOQA1035 pass1036 class TestServer(ServerSlice):1037 @protocol.handle(test_method)1038 async def test_methodY(self) -> ReturnValue[bytes]: # NOQA1039 return ReturnValue(response=byte_stream, content_type=OCTET_STREAM_CONTENT)1040 rs = Server()1041 server = TestServer(name="testserver")1042 rs.add_slice(server)1043 await rs.start()1044 async_finalizer.add(server.stop)1045 async_finalizer.add(rs.stop)1046 # client based calls1047 client = protocol.Client("client")1048 response = await client.test_method()1049 assert response.code == 2001050 assert response.result == byte_stream1051async def test_zip_content_type(unused_tcp_port, postgres_db, database_name, async_finalizer):1052 """Test whether API endpoints with an application/zip content-type work."""1053 configure(unused_tcp_port, database_name, postgres_db.port)1054 zip_content = b"test123"1055 @protocol.typedmethod(path="/test", operation="GET", client_types=["api"])1056 def test_method() -> ReturnValue[bytes]: # NOQA1057 pass1058 class TestServer(ServerSlice):1059 @protocol.handle(test_method)1060 async def test_methodY(self) -> ReturnValue[bytes]: # NOQA1061 return ReturnValue(response=zip_content, content_type=ZIP_CONTENT)1062 rs = Server()1063 server = TestServer(name="testserver")1064 rs.add_slice(server)1065 await rs.start()1066 async_finalizer.add(server.stop)1067 async_finalizer.add(rs.stop)1068 # client based calls1069 client = protocol.Client("client")1070 response = await client.test_method()1071 assert response.code == 2001072 assert response.result == zip_content1073@pytest.fixture1074async def options_server():1075 @protocol.typedmethod(path="/test", operation="OPTIONS", client_types=["api"])1076 def test_method() -> ReturnValue[str]: # NOQA1077 pass1078 class TestServer(ServerSlice):1079 @protocol.handle(test_method)1080 async def test_methodY(self) -> ReturnValue[str]: # NOQA1081 return ReturnValue(response="content")1082 return TestServer(name="testserver")1083@pytest.fixture1084def options_request(unused_tcp_port):1085 return HTTPRequest(1086 url=f"http://localhost:{unused_tcp_port}/api/v1/test",1087 method="OPTIONS",1088 connect_timeout=1.0,1089 request_timeout=1.0,1090 decompress_response=True,1091 )1092@pytest.mark.parametrize("auth_enabled, auth_header_allowed", [(True, True), (False, False)])1093async def test_auth_enabled_options_method(1094 auth_enabled,1095 auth_header_allowed,1096 unused_tcp_port,1097 postgres_db,1098 database_name,1099 async_finalizer,1100 options_server,1101 options_request,1102):1103 configure(unused_tcp_port, database_name, postgres_db.port)1104 config.Config.set("server", "auth", str(auth_enabled))1105 rs = Server()1106 rs.add_slice(options_server)1107 await rs.start()1108 async_finalizer.add(options_server.stop)1109 async_finalizer.add(rs.stop)1110 client = AsyncHTTPClient()1111 response = await client.fetch(options_request)1112 assert response.code == 2001113 assert ("Authorization" in response.headers.get("Access-Control-Allow-Headers")) == auth_header_allowed1114async def test_required_header_not_present(server):1115 client = AsyncHTTPClient()1116 response = await client.fetch(f"http://localhost:{server_bind_port.get()}/api/v2/environment_settings", raise_error=False)1117 assert response.code == 4001118async def test_malformed_json(server):1119 """1120 Tests sending malformed json to the server1121 """1122 port = opt.get_bind_port()1123 url = f"http://localhost:{port}/api/v2/environment"1124 request = HTTPRequest(url=url, method="PUT", body='{"name": env}')1125 client = AsyncHTTPClient()1126 response = await client.fetch(request, raise_error=False)1127 assert response.code == 4001128 assert (1129 json.loads(response.body)["message"]1130 == "The request body couldn't be decoded as a JSON: Expecting value: line 1 column 10 (char 9)"1131 )1132async def test_tuple_index_out_of_range(unused_tcp_port, postgres_db, database_name, async_finalizer):1133 configure(unused_tcp_port, database_name, postgres_db.port)1134 class Project(BaseModel):1135 name: str1136 value: str1137 class ProjectServer(ServerSlice):1138 @protocol.typedmethod(1139 api_prefix="test", path="/project/<project>", operation="GET", arg_options=ENV_OPTS, client_types=["api"]1140 )1141 def test_method(1142 tid: uuid.UUID, project: str, include_deleted: bool = False1143 ) -> List[Union[uuid.UUID, Project, bool]]: # NOQA1144 pass1145 @protocol.handle(test_method)1146 async def test_method(1147 tid: uuid.UUID, project: Project, include_deleted: bool = False1148 ) -> List[Union[uuid.UUID, Project, bool]]: # NOQA1149 return [tid, project, include_deleted]1150 rs = Server()1151 server = ProjectServer(name="projectserver")1152 rs.add_slice(server)1153 await rs.start()1154 async_finalizer.add(server.stop)1155 async_finalizer.add(rs.stop)1156 port = opt.get_bind_port()1157 url = f"http://localhost:{port}/test/v1/project/afcb51dc-1043-42b6-bb99-b4fc88603126"1158 request = HTTPRequest(url=url, method="GET")1159 client = AsyncHTTPClient()1160 response = await client.fetch(request, raise_error=False)1161 assert response.code == 4001162 assert json.loads(response.body)["message"] == "Invalid request: Field 'tid' is required."1163async def test_multiple_path_params(unused_tcp_port, postgres_db, database_name, async_finalizer):1164 configure(unused_tcp_port, database_name, postgres_db.port)1165 class ProjectServer(ServerSlice):1166 @protocol.typedmethod(path="/test/<id>/<name>", operation="GET", client_types=["api"])1167 def test_method(id: str, name: str, age: int) -> str: # NOQA1168 pass1169 @protocol.handle(test_method)1170 async def test_methodY(self, id: str, name: str, age: int) -> str: # NOQA1171 return name1172 rs = Server()1173 server = ProjectServer(name="projectserver")1174 rs.add_slice(server)1175 await rs.start()1176 async_finalizer.add(server.stop)1177 async_finalizer.add(rs.stop)1178 request = MethodProperties.methods["test_method"][0].build_call(args=[], kwargs={"id": "1", "name": "monty", "age": 42})1179 assert request.url == "/api/v1/test/1/monty?age=42"1180async def test_2151_method_header_parameter_in_body(async_finalizer) -> None:1181 async def _id(x: object, dct: Dict[str, str]) -> object:1182 return x1183 @protocol.method(1184 path="/testmethod",1185 operation="POST",1186 arg_options={"header_param": ArgOption(header="X-Inmanta-Header-Param", getter=_id)},1187 client_types=[const.ClientType.api],1188 )1189 def test_method(header_param: str, body_param: str) -> None:1190 """1191 A method used for testing.1192 """1193 class TestSlice(ServerSlice):1194 @protocol.handle(test_method)1195 async def test_method_implementation(self, header_param: str, body_param: str) -> None:1196 pass1197 server: Server = Server()1198 server_slice: ServerSlice = TestSlice("my_test_slice")1199 server.add_slice(server_slice)1200 await server.start()1201 async_finalizer.add(server_slice.stop)1202 async_finalizer.add(server.stop)1203 client = tornado.httpclient.AsyncHTTPClient()1204 # valid request should succeed1205 request = tornado.httpclient.HTTPRequest(1206 url=f"http://localhost:{opt.get_bind_port()}/api/v1/testmethod",1207 method="POST",1208 body=json_encode({"body_param": "body_param_value"}),1209 headers={"X-Inmanta-Header-Param": "header_param_value"},1210 )1211 response: tornado.httpclient.HTTPResponse = await client.fetch(request)1212 assert response.code == 2001213 # invalid request should fail1214 request = tornado.httpclient.HTTPRequest(1215 url=f"http://localhost:{opt.get_bind_port()}/api/v1/testmethod",1216 method="POST",1217 body=json_encode({"header_param": "header_param_value", "body_param": "body_param_value"}),1218 )1219 with pytest.raises(tornado.httpclient.HTTPClientError):1220 await client.fetch(request)1221@pytest.mark.parametrize("return_value,valid", [(1, True), (None, True), ("Hello World!", False)])1222async def test_2277_typedmethod_return_optional(async_finalizer, return_value: object, valid: bool) -> None:1223 @protocol.typedmethod(1224 path="/typedtestmethod",1225 operation="GET",1226 client_types=[const.ClientType.api],1227 api_version=1,1228 )1229 def test_method_typed() -> Optional[int]:1230 """1231 A typedmethod used for testing.1232 """1233 class TestSlice(ServerSlice):1234 @protocol.handle(test_method_typed)1235 async def test_method_typed_implementation(self) -> Optional[int]:1236 return return_value # type: ignore1237 server: Server = Server()1238 server_slice: ServerSlice = TestSlice("my_test_slice")1239 server.add_slice(server_slice)1240 await server.start()1241 async_finalizer.add(server_slice.stop)1242 async_finalizer.add(server.stop)1243 client: protocol.Client = protocol.Client("client")1244 response: Result = await client.test_method_typed()1245 if valid:1246 assert response.code == 2001247 assert response.result == {"data": return_value}1248 else:1249 assert response.code == 4001250def test_method_strict_exception() -> None:1251 with pytest.raises(InvalidMethodDefinition, match="Invalid type for argument arg: Any type is not allowed in strict mode"):1252 @protocol.typedmethod(path="/testmethod", operation="POST", client_types=[const.ClientType.api])1253 def test_method(arg: Any) -> None:1254 pass1255async def test_method_nonstrict_allowed(async_finalizer) -> None:1256 @protocol.typedmethod(path="/zipsingle", operation="POST", client_types=[const.ClientType.api], strict_typing=False)1257 def merge_dicts(one: Dict[str, Any], other: Dict[str, int], any_arg: Any) -> Dict[str, Any]:1258 """1259 Merge two dicts.1260 """1261 class TestSlice(ServerSlice):1262 @protocol.handle(merge_dicts)1263 async def merge_dicts_impl(self, one: Dict[str, Any], other: Dict[str, int], any_arg: Any) -> Dict[str, Any]:1264 return {**one, **other}1265 server: Server = Server()1266 server_slice: ServerSlice = TestSlice("my_test_slice")1267 server.add_slice(server_slice)1268 await server.start()1269 async_finalizer.add(server_slice.stop)1270 async_finalizer.add(server.stop)1271 client: protocol.Client = protocol.Client("client")1272 one: Dict[str, Any] = {"my": {"nested": {"keys": 42}}}1273 other: Dict[str, int] = {"single_level": 42}1274 response: Result = await client.merge_dicts(one, other, None)1275 assert response.code == 2001276 assert response.result == {"data": {**one, **other}}1277@pytest.mark.parametrize(1278 "param_type,param_value,expected_url",1279 [1280 (1281 Dict[str, str],1282 {"a": "b", "c": "d", ",&?=%": ",&?=%."},1283 "/api/v1/test/1/monty?filter.a=b&filter.c=d&filter.%2C%26%3F%3D%25=%2C%26%3F%3D%25.",1284 ),1285 (1286 Dict[str, List[str]],1287 {"a": ["b"], "c": ["d", "e"], "g": ["h"]},1288 "/api/v1/test/1/monty?filter.a=b&filter.c=d&filter.c=e&filter.g=h",1289 ),1290 (1291 Dict[str, List[str]],1292 {"a": ["b"], "c": ["d", "e"], ",&?=%": [",&?=%", "f"], ".g.h": ["i"]},1293 "/api/v1/test/1/monty?filter.a=b&filter.c=d&filter.c=e"1294 "&filter.%2C%26%3F%3D%25=%2C%26%3F%3D%25&filter.%2C%26%3F%3D%25=f&filter..g.h=i",1295 ),1296 (1297 List[str],1298 [1299 "a ",1300 "b,",1301 "c",1302 ],1303 "/api/v1/test/1/monty?filter=a+&filter=b%2C&filter=c",1304 ),1305 (1306 List[str],1307 ["a", "b", ",&?=%", "c", "."],1308 "/api/v1/test/1/monty?filter=a&filter=b&filter=%2C%26%3F%3D%25&filter=c&filter=.",1309 ),1310 (List[str], ["a ", "b", "c", ","], "/api/v1/test/1/monty?filter=a+&filter=b&filter=c&filter=%2C"),1311 ],1312)1313async def test_dict_list_get_roundtrip(1314 unused_tcp_port, postgres_db, database_name, async_finalizer, param_type, param_value, expected_url1315):1316 configure(unused_tcp_port, database_name, postgres_db.port)1317 class ProjectServer(ServerSlice):1318 @protocol.typedmethod(path="/test/<id>/<name>", operation="GET", client_types=["api"], strict_typing=False)1319 def test_method(id: str, name: str, filter: param_type) -> Any: # NOQA1320 pass1321 @protocol.handle(test_method)1322 async def test_method(self, id: str, name: str, filter: param_type) -> Any: # NOQA1323 return filter1324 rs = Server()1325 server = ProjectServer(name="projectserver")1326 rs.add_slice(server)1327 await rs.start()1328 async_finalizer.add(server.stop)1329 async_finalizer.add(rs.stop)1330 request = MethodProperties.methods["test_method"][0].build_call(1331 args=[], kwargs={"id": "1", "name": "monty", "filter": param_value}1332 )1333 assert request.url == expected_url1334 client: protocol.Client = protocol.Client("client")1335 response: Result = await client.test_method(1, "monty", filter=param_value)1336 assert response.code == 2001337 assert response.result["data"] == param_value1338async def test_dict_get_optional(unused_tcp_port, postgres_db, database_name, async_finalizer):1339 configure(unused_tcp_port, database_name, postgres_db.port)1340 class ProjectServer(ServerSlice):1341 @protocol.typedmethod(path="/test/<id>/<name>", operation="GET", client_types=["api"])1342 def test_method(id: str, name: str, filter: Optional[Dict[str, str]] = None) -> str: # NOQA1343 pass1344 @protocol.handle(test_method)1345 async def test_method(self, id: str, name: str, filter: Optional[Dict[str, str]] = None) -> str: # NOQA1346 return ",".join(filter.keys()) if filter is not None else ""1347 rs = Server()1348 server = ProjectServer(name="projectserver")1349 rs.add_slice(server)1350 await rs.start()1351 async_finalizer.add(server.stop)1352 async_finalizer.add(rs.stop)1353 request = MethodProperties.methods["test_method"][0].build_call(1354 args=[], kwargs={"id": "1", "name": "monty", "filter": {"a": "b"}}1355 )1356 assert request.url == "/api/v1/test/1/monty?filter.a=b"1357 client: protocol.Client = protocol.Client("client")1358 response: Result = await client.test_method(1, "monty", filter={"a": "b", "c": "d"})1359 assert response.code == 2001360 assert response.result["data"] == "a,c"1361 response: Result = await client.test_method(1, "monty")1362 assert response.code == 2001363 assert response.result["data"] == ""1364async def test_dict_list_nested_get_optional(unused_tcp_port, postgres_db, database_name, async_finalizer):1365 configure(unused_tcp_port, database_name, postgres_db.port)1366 class ProjectServer(ServerSlice):1367 @protocol.typedmethod(path="/test/<id>/<name>", operation="GET", client_types=["api"])1368 def test_method(id: str, name: str, filter: Optional[Dict[str, List[str]]] = None) -> str: # NOQA1369 pass1370 @protocol.handle(test_method)1371 async def test_method(self, id: str, name: str, filter: Optional[Dict[str, List[str]]] = None) -> str: # NOQA1372 return ",".join(filter.keys()) if filter is not None else ""1373 rs = Server()1374 server = ProjectServer(name="projectserver")1375 rs.add_slice(server)1376 await rs.start()1377 async_finalizer.add(server.stop)1378 async_finalizer.add(rs.stop)1379 request = MethodProperties.methods["test_method"][0].build_call(1380 args=[], kwargs={"id": "1", "name": "monty", "filter": {"a": ["b"]}}1381 )1382 assert request.url == "/api/v1/test/1/monty?filter.a=b"1383 client: protocol.Client = protocol.Client("client")1384 response: Result = await client.test_method(1, "monty", filter={"a": "b", "c": ["d", "e"]})1385 assert response.code == 2001386 assert response.result["data"] == "a,c"1387 response: Result = await client.test_method(1, "monty")1388 assert response.code == 2001389 assert response.result["data"] == ""1390@pytest.mark.parametrize(1391 "param_type,expected_error_message",1392 [1393 (1394 Dict[str, Dict[str, str]],1395 "nested dictionaries and union types for dictionary values are not supported for GET requests",1396 ),1397 (1398 Dict[str, Union[str, List[str]]],1399 "nested dictionaries and union types for dictionary values are not supported for GET requests",1400 ),1401 (List[Dict[str, str]], "lists of dictionaries and lists of lists are not supported for GET requests"),1402 (List[List[str]], "lists of dictionaries and lists of lists are not supported for GET requests"),1403 ],1404)1405async def test_dict_list_get_invalid(1406 unused_tcp_port, postgres_db, database_name, async_finalizer, param_type, expected_error_message1407):1408 configure(unused_tcp_port, database_name, postgres_db.port)1409 with pytest.raises(InvalidMethodDefinition) as e:1410 class ProjectServer(ServerSlice):1411 @protocol.typedmethod(path="/test/<id>/<name>", operation="GET", client_types=["api"])1412 def test_method(id: str, name: str, filter: param_type) -> str: # NOQA1413 pass1414 @protocol.handle(test_method)1415 async def test_method(self, id: str, name: str, filter: param_type) -> str: # NOQA1416 return ""1417 assert expected_error_message in str(e)1418async def test_list_get_optional(unused_tcp_port, postgres_db, database_name, async_finalizer):1419 configure(unused_tcp_port, database_name, postgres_db.port)1420 class ProjectServer(ServerSlice):1421 @protocol.typedmethod(path="/test/<id>/<name>", operation="GET", client_types=["api"])1422 def test_method(id: str, name: str, sort: Optional[List[int]] = None) -> str: # NOQA1423 pass1424 @protocol.typedmethod(path="/test_uuid/<id>", operation="GET", client_types=["api"])1425 def test_method_uuid(id: str, sort: Optional[List[uuid.UUID]] = None) -> str: # NOQA1426 pass1427 @protocol.handle(test_method)1428 async def test_method(self, id: str, name: str, sort: Optional[List[int]] = None) -> str: # NOQA1429 return str(sort) if sort else ""1430 @protocol.handle(test_method_uuid)1431 async def test_method_uuid(self, id: str, sort: Optional[List[uuid.UUID]] = None) -> str: # NOQA1432 return str(sort) if sort else ""1433 rs = Server()1434 server = ProjectServer(name="projectserver")1435 rs.add_slice(server)1436 await rs.start()1437 async_finalizer.add(server.stop)1438 async_finalizer.add(rs.stop)1439 request = MethodProperties.methods["test_method"][0].build_call(1440 args=[], kwargs={"id": "1", "name": "monty", "sort": [1, 2]}1441 )1442 assert request.url == "/api/v1/test/1/monty?sort=1&sort=2"1443 client: protocol.Client = protocol.Client("client")1444 response: Result = await client.test_method(1, "monty", sort=[1, 2])1445 assert response.code == 2001446 assert response.result["data"] == "[1, 2]"1447 response: Result = await client.test_method(1, "monty")1448 assert response.code == 2001449 assert response.result["data"] == ""1450 uuids = [uuid.uuid4(), uuid.uuid4()]1451 request = MethodProperties.methods["test_method_uuid"][0].build_call(args=[], kwargs={"id": "1", "sort": uuids})1452 assert request.url == f"/api/v1/test_uuid/1?sort={uuids[0]}&sort={uuids[1]}"1453async def test_dicts_multiple_get(unused_tcp_port, postgres_db, database_name, async_finalizer):1454 configure(unused_tcp_port, database_name, postgres_db.port)1455 class ProjectServer(ServerSlice):1456 @protocol.typedmethod(path="/test/<id>/<name>", operation="GET", client_types=["api"])1457 def test_method(id: str, name: str, filter: Dict[str, List[str]], another_filter: Dict[str, str]) -> str: # NOQA1458 pass1459 @protocol.handle(test_method)1460 async def test_method(1461 self, id: str, name: str, filter: Dict[str, List[str]], another_filter: Dict[str, str]1462 ) -> str: # NOQA1463 return ",".join(chain(filter.keys(), another_filter.keys()))1464 rs = Server()1465 server = ProjectServer(name="projectserver")1466 rs.add_slice(server)1467 await rs.start()1468 async_finalizer.add(server.stop)1469 async_finalizer.add(rs.stop)1470 request = MethodProperties.methods["test_method"][0].build_call(1471 args=[], kwargs={"id": "1", "name": "monty", "filter": {"a": ["b", "c"]}, "another_filter": {"d": "e"}}1472 )1473 assert request.url == "/api/v1/test/1/monty?filter.a=b&filter.a=c&another_filter.d=e"1474 client: protocol.Client = protocol.Client("client")1475 response: Result = await client.test_method(1, "monty", filter={"a": ["b"], "c": ["d", "e"]}, another_filter={"x": "y"})1476 assert response.code == 2001477 assert response.result["data"] == "a,c,x"1478async def test_dict_list_get_by_url(unused_tcp_port, postgres_db, database_name, async_finalizer):1479 configure(unused_tcp_port, database_name, postgres_db.port)1480 class ProjectServer(ServerSlice):1481 @protocol.typedmethod(path="/test/<id>/<name>", operation="GET", client_types=["api"])1482 def test_method(id: str, name: str, filter: Dict[str, str]) -> str: # NOQA1483 pass1484 @protocol.typedmethod(path="/test_list/<id>", operation="GET", client_types=["api"])1485 def test_method_list(id: str, filter: List[int]) -> str: # NOQA1486 pass1487 @protocol.typedmethod(path="/test_dict_of_lists/<id>", operation="GET", client_types=["api"])1488 def test_method_dict_of_lists(id: str, filter: Dict[str, List[str]]) -> str: # NOQA1489 pass1490 @protocol.handle(test_method)1491 async def test_method(self, id: str, name: str, filter: Dict[str, str]) -> str: # NOQA1492 return ",".join(filter.keys())1493 @protocol.handle(test_method_list)1494 async def test_method_list(self, id: str, filter: List[int]) -> str: # NOQA1495 return str(filter)1496 @protocol.handle(test_method_dict_of_lists)1497 async def test_method_dict_of_lists(self, id: str, filter: Dict[str, List[str]]) -> str: # NOQA1498 return ",".join(filter.keys())1499 rs = Server()1500 server = ProjectServer(name="projectserver")1501 rs.add_slice(server)1502 await rs.start()1503 async_finalizer.add(server.stop)1504 async_finalizer.add(rs.stop)1505 client = AsyncHTTPClient()1506 response = await client.fetch(f"http://localhost:{server_bind_port.get()}/api/v1/test/1/monty?filter.=b", raise_error=False)1507 assert response.code == 4001508 response = await client.fetch(1509 f"http://localhost:{server_bind_port.get()}/api/v1/test/1/monty?filter.a=b&filter.=c", raise_error=False1510 )1511 assert response.code == 4001512 response = await client.fetch(1513 f"http://localhost:{server_bind_port.get()}/api/v1/test/1/monty?filter.a=b&filter.a=c", raise_error=False1514 )1515 assert response.code == 4001516 response = await client.fetch(1517 f"http://localhost:{server_bind_port.get()}/api/v1/test_list/1?filter.a=b&filter.=c", raise_error=False1518 )1519 assert response.code == 4001520 # Integer should also work1521 response = await client.fetch(1522 f"http://localhost:{server_bind_port.get()}/api/v1/test_list/1?filter=42&filter=45", raise_error=False1523 )1524 assert response.code == 2001525 # list nested in dict1526 response = await client.fetch(1527 f"http://localhost:{server_bind_port.get()}/api/v1/test_dict_of_lists/1?filter.a=42&filter.a=55&filter.b=e",1528 raise_error=False,1529 )1530 assert response.code == 2001531 filter_with_comma = {"filter.a": "42,55,%2C70", "filter.b": "e"}1532 url = url_concat(f"http://localhost:{server_bind_port.get()}/api/v1/test_dict_of_lists/1", filter_with_comma)1533 response = await client.fetch(1534 url,1535 raise_error=False,1536 )1537 assert response.code == 2001538 response = await client.fetch(1539 f"http://localhost:{server_bind_port.get()}/api/v1/test_dict_of_lists/1?filter.a=42&filter.a=55&filter.&filter.c=a",1540 raise_error=False,1541 )1542 assert response.code == 4001543 filter_with_comma = {"filter.a": "b", "filter.c": "e", "filter.,&?=%": ",&?=%"}1544 url = url_concat(f"http://localhost:{server_bind_port.get()}/api/v1/test/1/monty", filter_with_comma)1545 response = await client.fetch(1546 url,1547 raise_error=False,1548 )1549 assert response.code == 2001550async def test_api_datetime_utc(unused_tcp_port, postgres_db, database_name, async_finalizer):1551 """1552 Test API input and output conversion for timestamps. Objects should be either timezone-aware or implicit UTC.1553 """1554 configure(unused_tcp_port, database_name, postgres_db.port)1555 timezone: datetime.timezone = datetime.timezone(datetime.timedelta(hours=2))1556 now: datetime.datetime = datetime.datetime.now().astimezone(timezone)1557 naive_utc: datetime.datetime = now.astimezone(datetime.timezone.utc).replace(tzinfo=None)1558 class ProjectServer(ServerSlice):1559 @protocol.typedmethod(path="/test", operation="GET", client_types=["api"])1560 def test_method(timestamp: datetime.datetime) -> List[datetime.datetime]:1561 pass1562 @protocol.handle(test_method)1563 async def test_method(self, timestamp: datetime.datetime) -> List[datetime.datetime]:1564 assert timestamp.tzinfo is not None1565 assert timestamp == now1566 return [1567 now,1568 now.astimezone(datetime.timezone.utc),1569 now.astimezone(datetime.timezone.utc).replace(tzinfo=None),1570 ]1571 rs = Server()1572 server = ProjectServer(name="projectserver")1573 rs.add_slice(server)1574 await rs.start()1575 async_finalizer.add(server.stop)1576 async_finalizer.add(rs.stop)1577 client: protocol.Client = protocol.Client("client")1578 response: Result = await client.test_method(timestamp=now)1579 assert response.code == 2001580 assert all(pydantic.parse_obj_as(datetime.datetime, timestamp) == naive_utc for timestamp in response.result["data"])1581 response: Result = await client.test_method(timestamp=now.astimezone(datetime.timezone.utc))1582 assert response.code == 2001583 response: Result = await client.test_method(timestamp=now.astimezone(datetime.timezone.utc).replace(tzinfo=None))1584 assert response.code == 2001585 response: Result = await client.test_method(timestamp=now.replace(tzinfo=None))1586 assert response.code == 5001587 # Test REST API without going through Python client1588 port = opt.get_bind_port()1589 client = AsyncHTTPClient()1590 async def request(timestamp: datetime.datetime) -> tornado.httpclient.HTTPResponse:1591 request = HTTPRequest(1592 url=(1593 f"http://localhost:{port}/api/v1/test?timestamp="1594 f"{urllib.parse.quote(timestamp.isoformat(timespec='microseconds'))}"1595 ),1596 method="GET",1597 )1598 return await client.fetch(request)1599 response = await request(now)1600 assert response.code == 2001601 response = await request(now.astimezone(datetime.timezone.utc).replace(tzinfo=None))1602 assert response.code == 2001603 with pytest.raises(tornado.httpclient.HTTPClientError):1604 response = await request(now.replace(tzinfo=None))1605async def test_dict_of_list(unused_tcp_port, postgres_db, database_name, async_finalizer):1606 """1607 Test API input and output conversion for timestamps. Objects should be either timezone-aware or implicit UTC.1608 """1609 configure(unused_tcp_port, database_name, postgres_db.port)1610 class APydanticType(BaseModel):1611 attr: int1612 class ProjectServer(ServerSlice):1613 @protocol.typedmethod(path="/test", operation="GET", client_types=[const.ClientType.api])1614 def test_method(id: str) -> Dict[str, List[APydanticType]]:1615 pass1616 @protocol.handle(test_method)1617 async def test_method(self, id: str) -> Dict[str, List[APydanticType]]:1618 return {id: [APydanticType(attr=1), APydanticType(attr=5)]}1619 rs = Server()1620 server = ProjectServer(name="projectserver")1621 rs.add_slice(server)1622 await rs.start()1623 async_finalizer.add(server.stop)1624 async_finalizer.add(rs.stop)1625 client: protocol.Client = protocol.Client("client")1626 result = await client.test_method(id="test")1627 assert result.code == 200, result.result["message"]1628 assert result.result["data"] == {"test": [{"attr": 1}, {"attr": 5}]}1629async def test_return_value_with_meta(unused_tcp_port, postgres_db, database_name, async_finalizer):1630 configure(unused_tcp_port, database_name, postgres_db.port)1631 class ProjectServer(ServerSlice):1632 @protocol.typedmethod(path="/test", operation="GET", client_types=["api"])1633 def test_method(with_warning: bool) -> ReturnValueWithMeta[str]: # NOQA1634 pass1635 @protocol.handle(test_method)1636 async def test_method(self, with_warning: bool) -> ReturnValueWithMeta: # NOQA1637 metadata = {"additionalInfo": f"Today's bitcoin exchange rate is: {(random.random() * 100000):.2f}$"}1638 result = ReturnValueWithMeta(response="abcd", metadata=metadata)1639 if with_warning:1640 result.add_warnings(["Warning message"])1641 return result1642 rs = Server()1643 server = ProjectServer(name="projectserver")1644 rs.add_slice(server)1645 await rs.start()1646 async_finalizer.add(server.stop)1647 async_finalizer.add(rs.stop)1648 client: protocol.Client = protocol.Client("client")1649 response = await client.test_method(False)1650 assert response.code == 2001651 assert response.result["data"] == "abcd"1652 assert response.result["metadata"].get("additionalInfo") is not None1653 assert response.result["metadata"].get("warnings") is None1654 response = await client.test_method(True)1655 assert response.code == 2001656 assert response.result["data"] == "abcd"1657 assert response.result["metadata"].get("additionalInfo") is not None1658 assert response.result["metadata"].get("warnings") is not None1659async def test_kwargs(unused_tcp_port, postgres_db, database_name, async_finalizer):1660 """1661 Test the use and validation of methods that use common.ReturnValue1662 """1663 configure(unused_tcp_port, database_name, postgres_db.port)1664 class ProjectServer(ServerSlice):1665 @protocol.typedmethod(path="/test", operation="POST", client_types=[ClientType.api], varkw=True)1666 def test_method(id: str, **kwargs: object) -> Dict[str, str]: # NOQA1667 """1668 Create a new project1669 """1670 @protocol.handle(test_method)1671 async def test_method(self, id: str, **kwargs: object) -> Dict[str, str]:1672 return {"name": str(kwargs["name"]), "value": str(kwargs["value"])}1673 rs = Server()1674 server = ProjectServer(name="projectserver")1675 rs.add_slice(server)1676 await rs.start()1677 async_finalizer.add(server.stop)1678 async_finalizer.add(rs.stop)1679 client = protocol.Client("client")1680 result = await client.test_method(id="test", name="test", value=True)1681 assert result.code == 2001682 assert result.result["data"]["name"] == "test"...
plugin.py
Source:plugin.py
...85 res = await gen_obj.__anext__()86 return res87 def finalizer():88 """Yield again, to finalize."""89 async def async_finalizer():90 try:91 await gen_obj.__anext__()92 except StopAsyncIteration:93 pass94 else:95 msg = "Async generator fixture didn't stop."96 msg += "Yield only once."97 raise ValueError(msg)98 loop.run_until_complete(async_finalizer())99 request.addfinalizer(finalizer)100 return loop.run_until_complete(setup())101 fixturedef.func = wrapper102 elif inspect.iscoroutinefunction(fixturedef.func):103 coro = fixturedef.func104 fixture_stripper = FixtureStripper(fixturedef)105 fixture_stripper.add(FixtureStripper.EVENT_LOOP)106 def wrapper(*args, **kwargs):107 loop = fixture_stripper.get_and_strip_from(FixtureStripper.EVENT_LOOP, kwargs)108 async def setup():109 res = await coro(*args, **kwargs)110 return res111 return loop.run_until_complete(setup())112 fixturedef.func = wrapper...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run pytest-asyncio automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
