Testing simpy

[adampaw...@gmail.com]

Hi,

I have two questions about testing:

1. What's the best way to "mock" a simpy environment?
2. Can you give me some advice on how to test processes which yield other processes?

Thanks!

[Stefan Scherfke]

Hi Adam,

Could you please provide more details what you want to do and why?
(E.g., why do you want to / need to mock the env?)

Cheers,
Stefan

> Thanks!
>
> --
> You received this message because you are subscribed to the Google Groups "python-simpy" group.
> To unsubscribe from this group and stop receiving emails from it, send an email to python-simpy...@googlegroups.com.
> To post to this group, send email to python...@googlegroups.com.
> To view this discussion on the web visit https://groups.google.com/d/msgid/python-simpy/a2af6dfd-2a85-48eb-badf-9796978b4010%40googlegroups.com.
> For more options, visit https://groups.google.com/d/optout.

[Adam Duran]

Hi Stefan,

I tried to narrow down the code to only the relevant information. The local
variables like values, schedules and delays are usually a result of some
business logic (and are passed to the processes as arguments).

I'm struggling with writing tests for these methods. I know there are a couple
of different issues here, but mostly I would like to test that a process
starts the correct processes (I thought about maybe using something like
assert_called_with from unittest.mock) and that a process is created at the
right time (given its parent processes). I'm sorry to bother you, but I'm a bit
lost:)

Thanks,
Adam


import simpy
env = simpy.Environment()


class Example1(object):

    def __init__(self):
        self.attribute = []
        self.example_2_objects = []

    def parent_process(self, env, example_3_object):
        env.process(self.child_process_1(env))
        yield env.process(example_3_object.child_process_2(env, self))
        env.process(self.child_process_3(env, example_3_object))

    def child_process_1(self, env):
        value = 2
        delay = 2
        for x in range(value):
            schedule = [1,2]
            for x in schedule:
                example_2_object = Example2()
                self.example_2_objects.append(example_2_object)
                env.process(example_2_object.grandchild_process_1(env, self))
            yield env.timeout(delay)

    def child_process_3(self, env, example_3_object):
        while True:
            if len(self.attribute) < 1:
                yield env.process(example_3_object.child_process_2(env, self))
            else:
                yield env.timeout(1)

    def grandchild_process_2(self, env):
        self.attribute.append('something')
        yield env.timeout(1)


class Example2(object):

    def __init__(self):
        self.attribute = []
        self.example_4_objects = []

    def grandchild_process_1(self, env, example_1_object):
        delay = 2
        schedule = [0,1]
        for x in schedule:
            yield env.timeout(delay)
            example_4_object = Example4()
            self.example_4_objects.append(example_4_object)
            example_4_object.great_grandchild_fuction_1(example_1_object, self)


class Example3(object):

    def child_process_2(self, env, example_1_object):
        delay = 1
        yield env.timeout(delay)
        env.process(example_1_object.grandchild_process_2(env))


class Example4(object):

    def great_grandchild_fuction_1(self, example_1_object, example_2_object):
        if len(example_1_object.attribute) > 0:
            something = example_1_object.attribute.pop(-1)
            example_2_object.attribute.append(something)


example_3_object = Example3()
example_1_object = Example1()
env.process(example_1_object.parent_process(env, example_3_object))
env.run(20)

On Tue, Oct 18, 2016 at 5:05 PM, Stefan Scherfke <ste...@sofa-rockers.org> wrote:

Hi Adam,

> Am 2016-10-17 um 23:27 schrieb adampaw...@gmail.com:
>
> Hi,
>
> I have two questions about testing:
>
> 1. What's the best way to "mock" a simpy environment?
> 2. Can you give me some advice on how to test processes which yield other processes?
>

Could you please provide more details what you want to do and why?
(E.g., why do you want to / need to mock the env?)

Cheers,
Stefan

> Thanks!

>
> --
> You received this message because you are subscribed to the Google Groups "python-simpy" group.

> To unsubscribe from this group and stop receiving emails from it, send an email to python-simpy+unsubscribe@googlegroups.com.

[adampaw...@gmail.com]

"""Hi Stefan,
I looked through the simpy tests and found the test_get_state and the
test_return_value from test_process.py very helpful. Could you take a look at
the tests for the functions below and tell me if I'm going in the right
direction?

I noticed that in the test_get_state you are using the pem_b process to monitor
the pem_a process and was wondering if instead you could run the environment
until the point in time when you want to test the pem_a process
(like in test_parent_process_ends_at_the_right_time below)?

I also wanted to ask if there is a better way to test if child processes were
started by the parent process other than testing its return value?
"""

import unittest
from simpy import Environment

def parent_process(env):
    yield env.timeout(1)
    children = []
    for i in range(2):
        children.append(env.process(child_process(env)))
    return children

def child_process(env):
    yield env.timeout(1)


class ExampleTest(unittest.TestCase):

    def setUp(self):
        self.env = Environment()

    def test_parent_process_starts_at_the_right_time(self):

        def check_process_is_alive(self, process):
            yield self.env.timeout(0)
            self.assertTrue(process.is_alive)

        process = self.env.process(parent_process(self.env))
        self.env.process(check_process_is_alive(self, process))
        self.env.run()
   
    def test_parent_process_ends_at_the_right_time(self):
        process = self.env.process(parent_process(self.env))

        self.env.run(1)
        self.assertTrue(process.is_alive)

        self.env.run(2)
        self.assertFalse(process.is_alive)

    def test_parent_process_starts_2_child_processes(self):
        process = self.env.process(parent_process(self.env))

        self.env.run(2)
        self.assertEqual(len(process.value), 2)

    def test_child_processes_start_at_the_right_time(self):
        process = self.env.process(parent_process(self.env))

        self.env.run(1)
        # try statement from test_unavailable_value in test_event.py
        try:
            process.value
            assert False, 'Expected an exception'
        except AttributeError as e:
            assert e.args[0].endswith('is not yet available')

        self.env.run(1.1)
        self.assertTrue(process.value[0].is_alive)

    def test_child_processes_end_at_the_right_time(self):
        process = self.env.process(parent_process(self.env))

        self.env.run(2)
        self.assertTrue(process.value[0].is_alive)
        self.assertTrue(process.value[1].is_alive)

        self.env.run(3)
        self.assertFalse(process.value[0].is_alive)
        self.assertFalse(process.value[1].is_alive)


if __name__ == '__main__':
    unittest.main()