KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _is_worker method in pytest-cov

Best Python code snippet using pytest-cov

account_data.py

Source:account_data.py

1# -*- coding: utf-8 -*-2# Copyright 2015, 2016 OpenMarket Ltd3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8#     http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15import logging16from synapse.api.errors import AuthError, NotFoundError, SynapseError17from synapse.http.servlet import RestServlet, parse_json_object_from_request18from ._base import client_patterns19logger = logging.getLogger(__name__)20class AccountDataServlet(RestServlet):21    """22    PUT /user/{user_id}/account_data/{account_dataType} HTTP/1.123    GET /user/{user_id}/account_data/{account_dataType} HTTP/1.124    """25    PATTERNS = client_patterns(26        "/user/(?P<user_id>[^/]*)/account_data/(?P<account_data_type>[^/]*)"27    )28    def __init__(self, hs):29        super(AccountDataServlet, self).__init__()30        self.auth = hs.get_auth()31        self.store = hs.get_datastore()32        self.notifier = hs.get_notifier()33        self._is_worker = hs.config.worker_app is not None34    async def on_PUT(self, request, user_id, account_data_type):35        if self._is_worker:36            raise Exception("Cannot handle PUT /account_data on worker")37        requester = await self.auth.get_user_by_req(request)38        if user_id != requester.user.to_string():39            raise AuthError(403, "Cannot add account data for other users.")40        body = parse_json_object_from_request(request)41        max_id = await self.store.add_account_data_for_user(42            user_id, account_data_type, body43        )44        self.notifier.on_new_event("account_data_key", max_id, users=[user_id])45        return 200, {}46    async def on_GET(self, request, user_id, account_data_type):47        requester = await self.auth.get_user_by_req(request)48        if user_id != requester.user.to_string():49            raise AuthError(403, "Cannot get account data for other users.")50        event = await self.store.get_global_account_data_by_type_for_user(51            account_data_type, user_id52        )53        if event is None:54            raise NotFoundError("Account data not found")55        return 200, event56class RoomAccountDataServlet(RestServlet):57    """58    PUT /user/{user_id}/rooms/{room_id}/account_data/{account_dataType} HTTP/1.159    GET /user/{user_id}/rooms/{room_id}/account_data/{account_dataType} HTTP/1.160    """61    PATTERNS = client_patterns(62        "/user/(?P<user_id>[^/]*)"63        "/rooms/(?P<room_id>[^/]*)"64        "/account_data/(?P<account_data_type>[^/]*)"65    )66    def __init__(self, hs):67        super(RoomAccountDataServlet, self).__init__()68        self.auth = hs.get_auth()69        self.store = hs.get_datastore()70        self.notifier = hs.get_notifier()71        self._is_worker = hs.config.worker_app is not None72    async def on_PUT(self, request, user_id, room_id, account_data_type):73        if self._is_worker:74            raise Exception("Cannot handle PUT /account_data on worker")75        requester = await self.auth.get_user_by_req(request)76        if user_id != requester.user.to_string():77            raise AuthError(403, "Cannot add account data for other users.")78        body = parse_json_object_from_request(request)79        if account_data_type == "m.fully_read":80            raise SynapseError(81                405,82                "Cannot set m.fully_read through this API."83                " Use /rooms/!roomId:server.name/read_markers",84            )85        max_id = await self.store.add_account_data_to_room(86            user_id, room_id, account_data_type, body87        )88        self.notifier.on_new_event("account_data_key", max_id, users=[user_id])89        return 200, {}90    async def on_GET(self, request, user_id, room_id, account_data_type):91        requester = await self.auth.get_user_by_req(request)92        if user_id != requester.user.to_string():93            raise AuthError(403, "Cannot get account data for other users.")94        event = await self.store.get_account_data_for_room_and_type(95            user_id, room_id, account_data_type96        )97        if event is None:98            raise NotFoundError("Room account data not found")99        return 200, event100def register_servlets(hs, http_server):101    AccountDataServlet(hs).register(http_server)...

conftest.py

Source:conftest.py

...9from clients.mysql_client import MysqlORMClient10from constants.app_constants import DbProperty11def pytest_addoption(parser):12    parser.addoption('--browser', default='chrome')13def _is_worker(config) -> bool:14    return hasattr(config, "workerinput")15def pytest_configure(config):16    mysql_orm_client = MysqlORMClient(17        user=DbProperty.DB_USER,18        password=DbProperty.DB_PASS,19        db_name=DbProperty.DB_NAME20    )21    base_dir = log_base_dir()22    if not _is_worker(config):23        wait_ready_app()24        mysql_orm_client.recreate_db()25    mysql_orm_client.connect(db_created=True)26    if not _is_worker(config):27        mysql_orm_client.prepare_create_tables()28        if os.path.exists(base_dir):29            shutil.rmtree(base_dir)30        os.makedirs(base_dir)31    config.base_temp_dir = base_dir32    config.mysql = mysql_orm_client33def pytest_unconfigure(config):34    if not _is_worker(config):35        utils.utils.run_command("chmod -R 777 /tmp/allure")36@pytest.fixture(scope='session')37def mysql_orm_client(request) -> MysqlORMClient:38    client = request.config.mysql39    yield client40    client.connection.close()41@pytest.fixture(scope='session')42def api_client():43    return ApiClient()44@pytest.fixture(scope='session')45def config(request) -> dict:46    browser = request.config.getoption('--browser')47    return {'browser': browser}48@pytest.fixture(scope='function')...

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.