getting the valid input in mock

[dvu4]

Hi,

I tried to mock the input type of a function create_service in services.py .

from mockito import verify, when, ANY, not_
from .api import services
from .models import Token


class TestCreateServicePrincipal(TestCase):
	def test_input_name_service_principal(self):
		when(services).create_service(ANY(Token), ANY(str), None, None, "p",False).thenReturn(1)
		when(services).create_service(not_(ANY(Token)), ANY(str), None, None, "p",False).thenRaise(TypeError)
		when(services).create_service(ANY(Token), not_( ANY(str)), None, None, "p",False).thenRaise(TypeError)
		when(services).create_service(not_(ANY(Token)), not_( ANY(str)), None, None, "p",False).thenRaise(TypeError)
		verify(services).create_service(ANY(Token), ANY(str), None, None, "p",False)
                #verify(services).create_service(not_(ANY(Token)), ANY(str), None, None, "p",False)
                #verify(services).create_service(ANY(Token), not_( ANY(str)), None, None, "p",False).thenRaise(TypeError)
		#verify(services).create_service(not_(ANY(Token)), not_( ANY(str)), None, None, "p",False).thenRaise(TypeError)

I am setting up a mock with when and then using verify to evaluate the valid or invalid inputs but I got the following error. Is there any way I can fix the error?

mockito.verification.VerificationError: Wanted but not invoked: create_service(<Any: <class 'Token'>>, <Any: <class 'str'>>, None, None, 'p', False) Instead got: Nothing

[kaste]

I think there is a misconception here. You first when something and then immediately verify the usage of the mock but you actually never used the mock. T.i. there is no execution phase in the test. (when = setup, verify = assert, the execution of the system under test is missing)

[dvu4]

Can you explain more what is missing in the execution of the system under test?

[kaste]

Typically

when(services).create_service(...).thenReturn(1)  # setup
assert services.create_service(1, 2, 3, 4, 5, 6) == 1  # usage
# now you could verify; you don't need though because you already asserted
unstub()  #  clean up (undo the stubbing) after the test, usually in the tearDown()

[dvu4]

Thank you for clarification. Is that any way I can assert it raises the correct exception like TypeError. I tried with invalid Token class

data = {
		"token_type": "mock_token", 
		"expires_in": 2023, 
		"ext_expires_in": 2025, 
		"access_token": 1 # --> valid one should be str , I changed it to int
	    }
token = Token(**data)
long_name = "foo"
when(services).create_service(not_(ANY(Token)), ANY(str), None, None, "p",False).thenRaise(TypeError)  # setup
assert services.create_service(token, long_name, None, None, "p",False)  # usage

But I got the error

mockito.invocation.InvocationError: 
Called but not expected:

    create_service(Token(token_type='mock_token', expires_in=2023, ext_expires_in=2025, access_token='1'), 'foo', None, None, 'p', False)

Stubbed invocations are:

    create_service(<Not: <Any: <class 'models.Token'>>>, <Any: <class 'str'>>, None, None, 'p', False)

[kaste]

I still think there is something odd here. Are you experienced with mock driven development or TDD etc. ?
The when(services).create_serceive(not_(any_()),...) seems untypical specific. The matchers not_(ANY(...))) translate to if not isinstance(token, Token: ... raise TypeError but then you actually call it with a token so this never matches the expected/stubbed invocation. If you define one very specific when invocation, mockito expects and only responds iff the system under tests calls and uses that exact same call/invocation signature. Otherwise it throws "unexpected invocation".

E.g.

when(os.path).exists("foo").thenReturn("No")
# usage
os.path.exists("bar")   # <= this will throw

For your code, typically the real implementation of create_service will throw TypeError if it is called with the wrong arguments. I assume you don't want to execute the code in create_service. How does the code look you actually test here? You don't test create_service here but the function which calls create_service. Am I right?

I see that the token you instantiate is an invalid token, but it is an instance of Token. Who decides this token is invalid?

[dvu4]

I am not experienced with mock driven development or TDD. My main goal is to test if the input is valid or not. The instance Token is defined below,

class Token(BaseModel):
    token_type: str
    expires_in: int
    ext_expires_in: int
    access_token: Optional[str] = None

I don't want to execute the code in create_service. I just want to throw exception such as TypeError if input is incorrect.
I am not sure if mockito allows me to define the correct type such as when(services).create_serceive(not_(any_()),...) , then test if I add a specific input to test the validity of the input.

I am wrong when calling token with a Token in The matchers not_(ANY(Token))), so I changed token to a random string

	def test_invalid_token_valid_long_name(self):
		token = "mock_token"
		long_name = "foo"

		when(service).create_service(not_(ANY(Token)), ANY(str), None, None, "p",False).thenRaise(TypeError)
		service.create_service(token,long_name, None, None, "p", False)

Is it normal if this error occur?

  File "/Users/.venv/lib/python3.9/site-packages/mockito/invocation.py", line 410, in raise_
    raise exception
TypeError

[kaste]

Well now you get the TypeError you want. However, I think the testing strategy should be different. We should focus on the function you're actually testing first.

Maybe it looks like:

def make_service(token, service):
   	try:
		return service.create_service(token, ...)
	except TypeError:
		return None

If this is the function under test, then one test could look like:

def test_returns_none_if_create_service_fails(self):
	when(service).create_service(...).thenRaise(TypeError)
 	assert make_service(mock(), service) is None

def test_passes_token_down(self):
	token = mock()
	expect(service).create_service(token, ...)
	make_service(token, service)
	verify(service).create_service(token, ...)

These are just examples. To help you out, you should focus and maybe post the function-under-test. This function uses a dependency (service) or a resource, or does a unwanted side-effect we want to capture. (Anything that is not a pure computation. Something that is not a mathematical function probably.)

Everything else, we also have arg_that, which lets you check an argument basically to program your own matcher, e.g.:

def check_token(token):
	if not isinstance(token.access_token, int): 
		return False
	return True
when(service).create_service(arg_that(check_token), ...).thenRaise(TypeError)
service.create_service(_some_invalid_token_)  # this should throw now

[dvu4]

The function I test is create_service in service.py which importing some external dependencies and may cause some unwanted side-effect

import  exclude_dict, service_names, call_api, 
from .models import AK,  SDetails, Token

class ServiceExists(Exception):
    pass

def create_service(
    token: Token,
    long_name: str,
    dn: Optional[str] = None,
    sn: Optional[str] = None,
    e: Optional[str] = None,
    t: Optional[str] = "p",
    d: bool = False
) -> Optional[AK]:

    if dn is None and (sn is None or e is None):
        raise ValueError(
            "You must set the `dn` or you must set both `sn and e`.")
    
    sp_list = service_names(token)

    if display_name in sp_list:
        raise ServiceExists(f"Service  {dn} exists, please create different name")


    sp_details = SDetails(
        dN=dn,
        tR=f"{long_name}",
        cS="D",
    )

    data = [exclude_dict(sp_details)]
    response = call_api(token, "post", "./service", json=data)
    result = parse_obj_as(AK, response)

    return result

I need to test all the valid inputs, output type is AK , all raise exception and correct message is showed if exception is raised and maybe call_api to see if it executed successfully or not.

[kaste]

Ok, when create_service is the system-under-test then you don't mock it out. You mock its dependencies. Of course create_service is already a complex function and it already has been written before the tests so this is a bit painful.

You look at the dependencies: call_api, service_names, parse_obj_as, exclude_dict. Typically some of them are pure (maybe: parse_obj_as) and could run in a test and some shouldn't really run in a test scenario (probably: call_api). Start with mocking the latter.

In test_create_service.py:

import service as sut   # sut = system-under-test
def test_happy_path(self):
    token, response, result = mock(), mock(), mock()
    when(sut).service_names(token).thenReturn([])
    when(sut).call_api(token, ...).thenReturn(response)   
    when(sut).parse_obj_as(sut.AK, response).thenReturn(result)

    assert sut.create_service(token, "long_name", "dn") == result

Make this test somehow pass.

It gets easier if the first happy path passes.

def test_no_dn_throws(self):
    token, response, result = mock(), mock(), mock()
    when(sut).service_names(token).thenReturn([])
    when(sut).call_api(token, ...).thenReturn(response)   
    when(sut).parse_obj_as(sut.AK, response).thenReturn(result)

    with self.assertRaises(ValueError):
        sut.create_service(token, "long_name")

And from there you test permutations and focus on edge cases and failing inputs. (There is probably already an error in the code as setting dn, sn, and e should throw.)

Does this gets you started?

[dvu4]

I got the error when running test_happy_path(self). Do I need to mock the SDetails and exclude_dict since the output of them will be used as json in call_api ?

Called but not expected:

    call_api(<Dummy id=34323>, 'post', /services', json=[{'app_Other_Type': 'long_name', 'calling_Service': 'DA', 'dN': 'dn'], debug=False)

Stubbed invocations are:

    call_api(<Dummy id=34323>, 'post', '/services', debug=False)

[kaste]

I thought SDetails and exclude_dict are "pure" and you can just fill out the json value manually. Isn't this what the function is about? Taking the arguments, and build a http request from them?

To get you started I actually wrote it with ... (the Ellipsis) -- call_api(token, ...) because in the first test you maybe don't care about the other arguments.

[dvu4]

Thank you for clarifying that for me. I filled out the json value manually and don't care about the arguments (...). I tried to test if the output of service_names is empty, then create a new service .

class AK(BaseModel):
    dM: Any
    code: Optional[int]
    message: str
    status: str

I want to confirm the output of test function, it works.

import service as sut   # sut = system-under-test

def test_happy_path(self):
        when(sut).service_names(...).thenReturn([])
	response = {
		"dM" : "foo",
    		"code" : 200,
    		"message":  "ok",
    		"status" : "200"
    		}

	when(sut).call_api(...).thenReturn(response)
        result = AK(**response)
        when(sut).parse_obj_as(AK, response).thenReturn(result)
	assert sut.create_service("token","long_name", "dn") == result

[dvu4]

I also tried to test if exception is raised, the correct message is showed. It works with pytest. Is there anyway I can do the same in mockito ?

class ServiceExists(Exception):
    Service already exists. Cannot create without the --force flag

import service as sut   # sut = system-under-test
import ServiceExists

def test_exception_message_existed_service(self):
        with pytest.raises(ServiceExists) as err:
	        with mock.patch('sut.service_names') as patched_function:
		        patched_function.return_value = ["ps", "rt", "vc", "ro"]
			sut.create_service("token_foo", "ln_foo", "rt", "sn_foo", "e_foo", "p", False)


	msg = "Service already exists. Cannot create without the --force flag"
	assert msg in str(err.value)
	assert err.type == ServiceExists

[kaste]

I don't see why there should be any difference in the last example:

when(sut).service_names(...).thenReturn(["rt"])
with pytest.raises(ServiceExists) as err:
	sut.create_service("token_foo", "ln_foo", "rt", "sn_foo", "e_foo", "p", False)
# ...

[dvu4]

That works. Thank you !!!

[kaste]

Are you sure you know what you're doing here? 😁

Settings() is throwing because its parent BaseSettings.__init__ requires arguments. That's what I'm reading here. Where do you updated Settings() so it reads from some file? And how is this related to mockito?

[dvu4]

class Settings(BaseSettings):
    client_id: str
    thumbprint: str
    private_key_file: str
    authority: str = "https://example"
    scope: List[str] = ["example"]

    class Config:
        env_file = "~/.env"

And~/.env is like this

PRIVATE_KEY_FILE="key.pem"
THUMBPRINT="grgeerhe"
CLIENT_ID="few"

so when I call config = Settings(), config will be updated from .env file. I hope there are some way I can mock client_id, thumbprint, private_key_file without reading .env file because there is no .env file at pipeline job level in azure devops

[kaste]

So you want to use a mock in production? Actually it looks like it's throwing on your computer and not on azure devops.

[dvu4]

mock is passed on my computer and the error is throwing on azure devops. I wonder if I can use mock in production.

[kaste]

I don't think you should use mock in production, that's not a good design. It still looks like it is not in production though. The tests fail when they run on azure.

It feels brittle to have a plain Settings() here (without any args) work in production because it in the end magically reads from somewhere. I think I would put them on the context which I then inject in the tests.

You can always do

when(main, strict=False).Settings().thenReturn(config)

to suppress the signature check (which gives you the TypeError).

I don't know why you put this call in the command. Settings is probably a singleton, so naturally you initiate it once and put it on module level ("global" but simple) and because that usually stinks you call it once and put it on the context. But you do and then strict=False might be what you're looking for 🤷

[dvu4]

I put Settings in the command because I need to get the access_token to create the service.
I also tried to initiate it in "global" but the error is throwing as soon as I import main, so calling it inside command, I can mock it and avoid the error on Azure.

Adding strict=False finally works because the error is not throwing any more 😄

[dvu4]

This test should be fail because 1234 is less than 10 characters but it still passed. I guess because of it always return msg

   def test_invalid_response(self):

        response = mock()
        access_token = mock()
        config = mock({"client_id": "str1", "thumbprint": "str2", "private_key_file": "str3"})
        msg = "App_Other_Type: size must be between 0 and 10"

        when(main, strict=False).Settings().thenReturn(config)
        when(main).get_token(config).thenReturn(access_token)
        when(main).create_service(access_token, ...).thenReturn(response)
        when(main).is_valid(response).thenReturn(False)
        when(main).show_message(response).thenReturn(msg)

        result = runner.invoke(app,
                               ["create", "1234", "--display-name", "p-prodfix-ds-pfoo-svcp"],
                               obj=mock({"debug": False}),
                               catch_exceptions=False)

        assert result.exit_code == 0
        assert msg in result.stdout

        unstub()

So I remove when(main).show_message(response).thenReturn(msg), then print(result.stdout) gives me

functools.partial(<function remembered_invocation_builder at 0x10369a710>, <mockito.mocking.Mock object at 0x104e2c100>, 'detailedMessage')
How can I access output of result to compare with msg ?

[kaste]

Idk what's going on here. If result.stdout is a patch you did a when() at some point. Interesting enough the mocked attribute name is detailedMessage

[kaste]

Btw you need to unstub on tearDown. Always possible that you don't have a clean environment otherwise. (In case you already have multiple tests.

[dvu4]

I unstub on teardown and it looks fine.

I tested for the valid case and the output of command is json and I mock the context but there is no output in result.stdout

def output_json(output: Optional[pathlib.Path], response: Optional[str]):
    if output and response:
        with open(output, 'w') as output_file:
            typer.echo(response.json(indent=2), file=output_file)
    elif response:
        typer.echo(response.json(indent=2))

@app.command()
def create(
    ctx: typer.Context,
    ln: str,
    dn: Optional[str] = None,
    sn: Optional[str] = None,
    e: Optional[str] = None,
    g: Optional[str] = "p",
    f: Optional[bool] = False,
) -> None:
    
    config = Settings()
    token = get_token(config) 
    response = create_service(token, ln, dn, sn, e, g, debug=ctx.obj.debug, f=f)
    if is_valid(response):
        output_json(ctx.obj.output, response)
    else:
        typer.echo(show_message(response))
        sys.exit(1)

 def test_valid_response(self):
        obj = mock({"output": "str"})
        context = mock({"obj": obj})

        response = mock()
        access_token = mock()
        config = mock({"client_id": "str1", "thumbprint": "str2", "private_key_file": "str3"})
        msg = "Successfully added Service Principal"

        output = {
            "detailedMessage": [],
            "httpCode": 201,
            "message": "Successfully added Service Principal",
            "status": "201 CREATED"}

        when(main, strict=False).Settings().thenReturn(config)
        when(main).get_token(config).thenReturn(access_token)
        when(main).create_service(access_token, ...).thenReturn(response)
        when(main).is_valid(response).thenReturn(True)
        when(main).output_json(context, response).thenReturn(output)

        result = runner.invoke(app,
                               ["create", "long_name", "--display-name", "p-prodfix-ds-pfoo-svcp"],
                               obj=mock({"debug": False}),
                               catch_exceptions=False)

        assert result.exit_code == 0
        assert msg in result.stdout

Traceback (most recent call last):
  File "/Users/tests/test_cli.py", line 69, in test_valid_response
    assert msg in result.stdout
AssertionError

[kaste]

The code you posted just doesn't match. you mock output_jsonwith ctx, response but you call it with ctx.obj.output, response. It is not super clear that you actually write something to stdout because you mock output_json (so it doesn't run).

[dvu4]

I tried with output_json(obj, response) where obj = mock({"output": "str"}) but it does not write something to stdout. I am not sure if I need to mock output_json

def test_valid_response(self):
        obj = mock({"output": "str"})
       
        response = mock()
        access_token = mock()
        config = mock({"client_id": "str1", "thumbprint": "str2", "private_key_file": "str3"})
        msg = "Successfully added Service Principal"

        output = {
            "detailedMessage": [],
            "httpCode": 201,
            "message": "Successfully added Service Principal",
            "status": "201 CREATED"}

        when(main, strict=False).Settings().thenReturn(config)
        when(main).get_token(config).thenReturn(access_token)
        when(main).create_service(access_token, ...).thenReturn(response)
        when(main).is_valid(response).thenReturn(True)
        when(main).output_json(obj, response).thenReturn(output)

        result = runner.invoke(app,
                               ["create", "long_name", "--display-name", "p-prodfix-ds-pfoo-svcp"],
                               obj=mock({"debug": False}),
                               catch_exceptions=False)

        assert result.exit_code == 0
        assert msg in result.stdout

[kaste]

When you patch output_json it does not run so it also doesn't write anything to stdout.

[dvu4]

is there any way I can get the json output when I do not patch output_json ?

[dvu4]

Hi,
I try to mock an instance of DeviceCodeCredential which initiates the device code authorization flow with Azure Active Directory and an instance SecretClient for testing the exception ServiceRequestError. I used credential = mock() and client = mock() but it did not work

services.py

from azure.identity import DeviceCodeCredential
from azure.keyvault.secrets import SecretClient
from azure.core.exceptions import ServiceRequestError, HttpResponseError


def add_keyvault(
    token: Token,
    keyvault_name: str,
    response: Response,
    dname: Optional[str] = None,
    appid: Optional[str] = None,
) -> None:

    credential = DeviceCodeCredential()
    kv_uri = f"https://{keyvault_name}.vault.azure.net"
    client = SecretClient(vault_url=kv_uri, credential=credential)

    sp_detail = list_service(token, appid=appid, dname=dname)

    dname = sp_detail.azureSPDetails[0].appName
    client_id = sp_detail.azureSPDetails[0].appId
    client_id_name = f"{dname}-client-id"

    secret_value = response.secret["secretText"]
    secret_name = f"{dname}-secret"

    try:
        set_client = client.set_secret(client_id_name, client_id)
        set_secret = client.set_secret(secret_name, secret_value)
    except ServiceRequestError as non_exist_keyvault_error:
        print(non_exist_keyvault_error.message)
        sys.exit(1)
    except HttpResponseError as non_permission_keyvault_error:
        print(non_permission_keyvault_error.message)
        sys.exit(1)

The test function :

    def test_keyvault_does_not_exit(self):
        token = mock()
        sp_detail = mock()
        when(services).list_service(token, ...).thenReturn(sp_detail)
        keyvault_name = "keyvault_foo"
        kv_uri = f"https://{keyvault_name}.vault.azure.net"
        credential = mock()
        client = mock()
        when(services).SecretClient(vault_url=kv_uri, credential=credential).thenReturn(client)
        response = {
            "detailedMessage": "Successfully added secret",
            "httpCode": 200,
            "message": "Successfully added secret",
            "secret": "12345",
            "status": "200 OK"
        }

        with pytest.raises(ServiceRequestError) as err:
            services.add_keyvault(
                token,
                keyvault_name,
                response,
                "display_name",
                "app_object_id"
            )

        msg = "Failed to establish a new connection"
        assert msg in str(err.value)
        assert err.type == ServiceRequestError
        unstub()

But I got the error. Is there any way I can mock this instance ?

    def add_keyvault(
        token: Token,
        keyvault_name: str,
        response: Response,
        dname: Optional[str] = None,
        appid: Optional[str] = None,
    ) -> None:
    
        credential = DeviceCodeCredential()
        kv_uri = f"https://{keyvault_name}.vault.azure.net"
>       client = SecretClient(vault_url=kv_uri, credential=credential)
E       mockito.invocation.InvocationError: 
E       Called but not expected:
E       
E           SecretClient(vault_url='https://keyvault_foo.vault.azure.net', credential=<azure.identity._credentials.device_code.DeviceCodeCredential object at 0x1068b6140>)
E       
E       Stubbed invocations are:
E       
E           SecretClient(vault_url='https://keyvault_foo.vault.azure.net', credential=<Dummy id=4404764512>)

src/services.py:331: InvocationError

[dvu4]

I want to verify the meaningful behavior of add_keyvault when keyvault_name is not available, then it should verify the message is correct. Or any side effect I am missing?

[kaste]

How to test stdout (print) in pytest https://docs.pytest.org/en/7.1.x/how-to/capture-stdout-stderr.html#accessing-captured-output-from-a-test-function

[dvu4]

I can test the stdout of the function app.py by injecting fixtures as instance attributes

def app(token):
     response = call_api(token,"get", f"/https")
     status_code = response.get("httpCode")
     msg = response.get("message")

    if status_code < 200 or status_code > 299:
           logging.error(msg, exc_info=True)

test.py

Class TestApp(TestCase):
    @pytest.fixture(autouse=True)
    def inject_fixtures(self, caplog):
        self._caplog = caplog

    def test_app_response(self):
        response = {
            'detailedMessage': 'name already exists',
            'httpCode': 400,
            'status': '400 BAD_REQUEST'}

        msg = 'name already exists'
        with self._caplog.at_level(logging.INFO):
            sut.app(token)
            assert msg in self._caplog.records[0].message
        unstub()

But for when(client).set_client("client_id_name", "client_id").thenRaise(ServiceRequestError), I can not get any stdout of add_keyvault. Maybe it could not see the message thrown by the exception

[kaste]

Don't know really. But when you do thenRaise(ServiceRequestError) there is really no error message, you need to thenRaise(ServiceRequestError("here comes my message")). Don't know if that is related. But that's also not stdout testing (capsys), you switched from the simple print("...") to logging.error it seems.

[dvu4]

I found that logging.errror is more acceptable when I build api module. print and sys.exit(1) is not.