From c2d4e69c353ddb65d7e060d350ffeb56060bb68a Mon Sep 17 00:00:00 2001 From: Chris Lenk Date: Fri, 21 Apr 2023 12:01:45 -0400 Subject: [PATCH] Use our own AsyncMock The one from unittest doesn't exist in Python 3.7. Also consolidate a second async mock wrapper for code reuse. --- tests/conftest.py | 2 +- tests/services/test_planning_svc.py | 12 +++--------- 2 files changed, 4 insertions(+), 10 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 6630c8a21..c968b8e91 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,7 +2,6 @@ import os.path import pytest -from unittest.mock import AsyncMock import random import string import uuid @@ -63,6 +62,7 @@ from app.api.rest_api import RestApi from app import version +from tests import AsyncMock DIR = os.path.dirname(os.path.abspath(__file__)) CONFIG_DIR = os.path.join(DIR, '..', 'conf') diff --git a/tests/services/test_planning_svc.py b/tests/services/test_planning_svc.py index d9369c5aa..655b3de16 100644 --- a/tests/services/test_planning_svc.py +++ b/tests/services/test_planning_svc.py @@ -11,6 +11,7 @@ from app.objects.secondclass.c_fact import Fact from app.objects.secondclass.c_requirement import Requirement from app.utility.base_world import BaseWorld +from tests import AsyncMock stop_bucket_exhaustion_params = [ @@ -69,13 +70,6 @@ def __init__(self, **kwargs): return PlannerStub(**kwargs) -def async_wrapper(return_value): - """Creates an async method that returns a constant value for mocking purposes.""" - async def wrap(*args, **kwargs): - return return_value - return wrap - - @pytest.fixture async def setup_planning_test(executor, ability, agent, operation, data_svc, event_svc, init_base_world): texecutor = executor(name='sh', platform='darwin', command='mkdir test', cleanup='rm -rf test') @@ -355,9 +349,9 @@ async def test_trim_links(self, setup_planning_test, planning_svc): Fact(trait='a.b.e', value='3'), ] - operation.all_facts = async_wrapper(facts) + operation.all_facts = AsyncMock(return_value=facts) operation.planner = MagicMock() - planning_svc.load_module = async_wrapper(RequirementFake()) + planning_svc.load_module = AsyncMock(return_value=RequirementFake()) link.ability.requirements = [Requirement('fake_requirement', [{'fake': 'relationship'}])] trimmed_links = await planning_svc.trim_links(operation, [link], agent)