import unittest
import time

from scanpointgenerator import LineGenerator, CompoundGenerator

from malcolm.core import Process, Part, Context, AlarmStatus, \
    AlarmSeverity, AbortedError
from malcolm.modules.scanning.parts import RunnableChildPart
from malcolm.modules.demo.blocks import ticker_block
from malcolm.compat import OrderedDict
from malcolm.modules.scanning.controllers import \
    RunnableController
from malcolm.modules.scanning.util import RunnableStates


class TestRunnableStates(unittest.TestCase):

    def setUp(self):
        self.o = RunnableStates()

    def test_init(self):
        expected = OrderedDict()
        expected['Resetting'] = {"Ready", "Fault", "Disabling"}
        expected['Ready'] = {"Configuring", "Aborting", 'Saving', "Fault",
                             "Disabling", "Loading"}
        expected['Saving'] = {'Fault', 'Ready', 'Disabling'}
        expected['Loading'] = {'Disabling', 'Fault', 'Ready'}
        expected['Configuring'] = {"Armed", "Aborting", "Fault", "Disabling"}
        expected['Armed'] = {"Seeking", "Aborting", "Running",
                             "Fault", "Disabling", "Resetting"}
        expected['Running'] = {"PostRun", "Seeking", "Aborting", "Fault",
                               "Disabling"}
        expected['PostRun'] = {"Ready", "Armed", "Aborting", "Fault",
                               "Disabling"}
        expected['Seeking'] = {"Armed", "Paused", "Aborting", "Fault",
                               "Disabling"}
        expected['Paused'] = {"Seeking", "Running", "Aborting", "Fault",
                              "Disabling"}
        expected['Aborting'] = {"Aborted", "Fault", "Disabling"}
        expected['Aborted'] = {"Resetting", "Fault", "Disabling"}
        expected['Fault'] = {"Resetting", "Disabling"}
        expected['Disabling'] = {"Disabled", "Fault"}
        expected['Disabled'] = {"Resetting"}
        assert self.o._allowed == expected
        possible_states = [
            'Ready', 'Resetting', 'Saving', 'Loading', 'Configuring', 'Armed',
            'Running', 'Seeking', 'PostRun', 'Paused', 'Aborting', 'Aborted',
            'Fault', 'Disabling', 'Disabled']
        assert self.o.possible_states == possible_states


class TestRunnableController(unittest.TestCase):
    def setUp(self):
        self.p = Process('process1')
        self.context = Context(self.p)

        # Make a ticker_block block to act as our child
        for c in ticker_block(mri="childBlock", config_dir="/tmp"):
            self.p.add_controller(c)
        self.b_child = self.context.block_view("childBlock")

        # Make an empty part for our parent
        part1 = Part("part1")

        # Make a RunnableChildPart to control the ticker_block
        part2 = RunnableChildPart(
            mri='childBlock', name='part2', initial_visibility=True)

        # create a root block for the RunnableController block to reside in
        self.c = RunnableController(mri='mainBlock', config_dir="/tmp")
        self.c.add_part(part1)
        self.c.add_part(part2)
        self.p.add_controller(self.c)
        self.b = self.context.block_view("mainBlock")
        self.ss = self.c.state_set

        # start the process off
        self.checkState(self.ss.DISABLED)
        self.p.start()
        self.checkState(self.ss.READY)

    def tearDown(self):
        self.p.stop(timeout=1)

    def checkState(self, state, child=True, parent=True):
        if child:
            assert self.b_child.state.value == state
        if parent:
            assert self.c.state.value == state

    def checkSteps(self, configured, completed, total):
        assert self.b.configuredSteps.value == configured
        assert self.b.completedSteps.value == completed
        assert self.b.totalSteps.value == total
        assert self.b_child.configuredSteps.value == configured
        assert self.b_child.completedSteps.value == completed
        assert self.b_child.totalSteps.value == total

    def test_init(self):
        assert self.c.completed_steps.value == 0
        assert self.c.configured_steps.value == 0
        assert self.c.total_steps.value == 0
        assert list(self.b.configure.takes.elements) == \
               ["generator", "axesToMove", "exceptionStep"]

    def test_reset(self):
        self.c.disable()
        self.checkState(self.ss.DISABLED)
        self.c.reset()
        self.checkState(self.ss.READY)

    def test_modify_child(self):
        # Save an initial setting for the child
        self.b_child.save("init_child")
        assert self.b_child.modified.value is False
        x = self.context.block_view("COUNTERX")
        x.counter.put_value(31)
        # x counter now at 31, child should be modified
        assert x.counter.value == 31
        assert self.b_child.modified.value is True
        assert self.b_child.modified.alarm.severity == AlarmSeverity.MINOR_ALARM
        assert self.b_child.modified.alarm.status == AlarmStatus.CONF_STATUS
        assert self.b_child.modified.alarm.message == \
            "x.counter.value = 31.0 not 0.0"
        self.prepare_half_run()
        self.b.run()
        # x counter now at 2, child should be modified by us
        assert self.b_child.modified.value is True
        assert self.b_child.modified.alarm.severity == AlarmSeverity.NO_ALARM
        assert self.b_child.modified.alarm.status == AlarmStatus.CONF_STATUS
        assert self.b_child.modified.alarm.message == \
            "(We modified) x.counter.value = 2.0 not 0.0"
        assert x.counter.value == 2.0
        x.counter.put_value(0.0)
        # x counter now at 0, child should be unmodified
        assert x.counter.value == 0
        assert self.b_child.modified.alarm.message == ""
        assert self.b_child.modified.value is False

    def test_modify_parent(self):
        # Save an initial setting for child and parent
        self.b_child.save("init_child")
        self.b.save("init_parent")
        # Change a value and save as a new child setting
        x = self.context.block_view("COUNTERX")
        x.counter.put_value(31)
        self.b_child.save("new_child")
        assert self.b_child.modified.value is False
        assert self.b.modified.value is True
        assert self.b.modified.alarm.severity == AlarmSeverity.MINOR_ALARM
        assert self.b.modified.alarm.status == AlarmStatus.CONF_STATUS
        assert self.b.modified.alarm.message == \
            "part2.design.value = 'new_child' not 'init_child'"
        # Load the child again
        self.b_child.design.put_value("new_child")
        assert self.b.modified.value is True
        # And check that loading parent resets it
        self.b.design.put_value("init_parent")
        assert self.b.modified.value is False
        assert self.b_child.design.value == "init_child"
        # Put back
        self.b_child.design.put_value("new_child")
        assert self.b.modified.value is True
        # Do a configure, and check we get set back
        self.prepare_half_run()
        assert self.b_child.design.value == "init_child"
        assert self.b_child.modified.value is False
        assert self.b.modified.value is False

    def test_abort(self):
        self.b.abort()
        self.checkState(self.ss.ABORTED)

    def test_validate(self):
        line1 = LineGenerator('y', 'mm', 0, 2, 3)
        line2 = LineGenerator('x', 'mm', 0, 2, 2)
        compound = CompoundGenerator([line1, line2], [], [])
        actual = self.b.validate(generator=compound, axesToMove=['x'])
        assert actual["generator"].to_dict() == compound.to_dict()
        assert actual["axesToMove"] == ['x']

    def prepare_half_run(self, duration=0.01, exception=0):
        line1 = LineGenerator('y', 'mm', 0, 2, 3)
        line2 = LineGenerator('x', 'mm', 0, 2, 2)
        compound = CompoundGenerator([line1, line2], [], [], duration)
        self.b.configure(
            generator=compound, axesToMove=['x'], exceptionStep=exception)

    def test_configure_run(self):
        assert self.b.configure.writeable is True
        assert self.b.configure.takes.elements["generator"].writeable is True
        assert self.b.validate.takes.elements["generator"].writeable is True
        assert self.b.validate.returns.elements["generator"].writeable is False
        self.prepare_half_run()
        self.checkSteps(2, 0, 6)
        self.checkState(self.ss.ARMED)
        assert self.b.configure.writeable is False
        assert self.b.configure.takes.elements["generator"].writeable is True
        assert self.b.validate.takes.elements["generator"].writeable is True
        assert self.b.validate.returns.elements["generator"].writeable is False

        self.b.run()
        self.checkState(self.ss.ARMED)
        self.checkSteps(4, 2, 6)

        self.b.run()
        self.checkState(self.ss.ARMED)
        self.checkSteps(6, 4, 6)

        self.b.run()
        self.checkState(self.ss.READY)

    def test_abort(self):
        self.prepare_half_run()
        self.b.run()
        self.b.abort()
        self.checkState(self.ss.ABORTED)

    def test_pause_seek_resume(self):
        self.prepare_half_run()
        self.checkSteps(configured=2, completed=0, total=6)
        self.b.run()
        self.checkState(self.ss.ARMED)
        self.checkSteps(4, 2, 6)
        self.b.pause(completedSteps=1)
        self.checkState(self.ss.ARMED)
        self.checkSteps(2, 1, 6)
        self.b.run()
        self.checkSteps(4, 2, 6)
        self.b.completedSteps.put_value(5)
        self.checkSteps(6, 5, 6)
        self.b.run()
        self.checkState(self.ss.READY)

    def test_resume_in_run(self):
        self.prepare_half_run(duration=0.5)
        f = self.b.run_async()
        self.context.sleep(0.95)
        self.b.pause()
        self.checkState(self.ss.PAUSED)
        self.checkSteps(2, 1, 6)
        self.b.resume()
        # Parent should be running, child won't have got request yet
        then = time.time()
        self.checkState(self.ss.RUNNING, child=False)
        self.context.wait_all_futures(f, timeout=2)
        now = time.time()
        self.checkState(self.ss.ARMED)
        self.checkSteps(4, 2, 6)
        # This test fails on Travis sometimes, looks like the docker container
        # just gets starved
        # self.assertAlmostEqual(now - then, 0.5, delta=0.1)

    def test_run_exception(self):
        self.prepare_half_run(exception=1)
        with self.assertRaises(AssertionError):
            self.b.run()
        self.checkState(self.ss.FAULT)

    def test_run_stop(self):
        self.prepare_half_run(duration=0.1)
        f = self.b.run_async()
        self.context.sleep(0.1)
        self.b.abort()
        with self.assertRaises(AbortedError):
            f.result()
        self.checkState(self.ss.ABORTED)


import unittest
import time

from scanpointgenerator import LineGenerator, CompoundGenerator

from malcolm.core import Process, Part, Context, AlarmStatus, \
    AlarmSeverity, AbortedError
from malcolm.modules.scanning.parts import RunnableChildPart
from malcolm.modules.demo.blocks import ticker_block
from malcolm.compat import OrderedDict
from malcolm.modules.scanning.controllers import \
    RunnableController
from malcolm.modules.scanning.util import RunnableStates


class TestRunnableStates(unittest.TestCase):

    def setUp(self):
        self.o = RunnableStates()

    def test_init(self):
        expected = OrderedDict()
        expected['Resetting'] = {"Ready", "Fault", "Disabling"}
        expected['Ready'] = {"Configuring", "Aborting", 'Saving', "Fault",
                             "Disabling", "Loading"}
        expected['Saving'] = {'Fault', 'Ready', 'Disabling'}
        expected['Loading'] = {'Disabling', 'Fault', 'Ready'}
        expected['Configuring'] = {"Armed", "Aborting", "Fault", "Disabling"}
        expected['Armed'] = {"Seeking", "Aborting", "Running",
                             "Fault", "Disabling", "Resetting"}
        expected['Running'] = {"PostRun", "Seeking", "Aborting", "Fault",
                               "Disabling"}
        expected['PostRun'] = {"Ready", "Armed", "Aborting", "Fault",
                               "Disabling"}
        expected['Seeking'] = {"Armed", "Paused", "Aborting", "Fault",
                               "Disabling"}
        expected['Paused'] = {"Seeking", "Running", "Aborting", "Fault",
                              "Disabling"}
        expected['Aborting'] = {"Aborted", "Fault", "Disabling"}
        expected['Aborted'] = {"Resetting", "Fault", "Disabling"}
        expected['Fault'] = {"Resetting", "Disabling"}
        expected['Disabling'] = {"Disabled", "Fault"}
        expected['Disabled'] = {"Resetting"}
        assert self.o._allowed == expected
        possible_states = [
            'Ready', 'Resetting', 'Saving', 'Loading', 'Configuring', 'Armed',
            'Running', 'Seeking', 'PostRun', 'Paused', 'Aborting', 'Aborted',
            'Fault', 'Disabling', 'Disabled']
        assert self.o.possible_states == possible_states


class TestRunnableController(unittest.TestCase):
    def setUp(self):
        self.p = Process('process1')
        self.context = Context(self.p)

        # Make a ticker_block block to act as our child
        for c in ticker_block(mri="childBlock", config_dir="/tmp"):
            self.p.add_controller(c)
        self.b_child = self.context.block_view("childBlock")

        # Make an empty part for our parent
        part1 = Part("part1")

        # Make a RunnableChildPart to control the ticker_block
        part2 = RunnableChildPart(
            mri='childBlock', name='part2', initial_visibility=True)

        # create a root block for the RunnableController block to reside in
        self.c = RunnableController(mri='mainBlock', config_dir="/tmp")
        self.c.add_part(part1)
        self.c.add_part(part2)
        self.p.add_controller(self.c)
        self.b = self.context.block_view("mainBlock")
        self.ss = self.c.state_set

        # start the process off
        self.checkState(self.ss.DISABLED)
        self.p.start()
        self.checkState(self.ss.READY)

    def tearDown(self):
        self.p.stop(timeout=1)

    def checkState(self, state, child=True, parent=True):
        if child:
            assert self.b_child.state.value == state
        if parent:
            assert self.c.state.value == state

    def checkSteps(self, configured, completed, total):
        assert self.b.configuredSteps.value == configured
        assert self.b.completedSteps.value == completed
        assert self.b.totalSteps.value == total
        assert self.b_child.configuredSteps.value == configured
        assert self.b_child.completedSteps.value == completed
        assert self.b_child.totalSteps.value == total

    def test_init(self):
        assert self.c.completed_steps.value == 0
        assert self.c.configured_steps.value == 0
        assert self.c.total_steps.value == 0
        assert list(self.b.configure.takes.elements) == \
               ["generator", "axesToMove", "exceptionStep"]

    def test_reset(self):
        self.c.disable()
        self.checkState(self.ss.DISABLED)
        self.c.reset()
        self.checkState(self.ss.READY)

    def test_modify_child(self):
        # Save an initial setting for the child
        self.b_child.save("init_child")
        assert self.b_child.modified.value is False
        x = self.context.block_view("COUNTERX")
        x.counter.put_value(31)
        # x counter now at 31, child should be modified
        assert x.counter.value == 31
        assert self.b_child.modified.value is True
        assert self.b_child.modified.alarm.severity == AlarmSeverity.MINOR_ALARM
        assert self.b_child.modified.alarm.status == AlarmStatus.CONF_STATUS
        assert self.b_child.modified.alarm.message == \
            "x.counter.value = 31.0 not 0.0"
        self.prepare_half_run()
        self.b.run()
        # x counter now at 2, child should be modified by us
        assert self.b_child.modified.value is True
        assert self.b_child.modified.alarm.severity == AlarmSeverity.NO_ALARM
        assert self.b_child.modified.alarm.status == AlarmStatus.CONF_STATUS
        assert self.b_child.modified.alarm.message == \
            "(We modified) x.counter.value = 2.0 not 0.0"
        assert x.counter.value == 2.0
        x.counter.put_value(0.0)
        # x counter now at 0, child should be unmodified
        assert x.counter.value == 0
        assert self.b_child.modified.alarm.message == ""
        assert self.b_child.modified.value is False

    def test_modify_parent(self):
        # Save an initial setting for child and parent
        self.b_child.save("init_child")
        self.b.save("init_parent")
        # Change a value and save as a new child setting
        x = self.context.block_view("COUNTERX")
        x.counter.put_value(31)
        self.b_child.save("new_child")
        assert self.b_child.modified.value is False
        assert self.b.modified.value is True
        assert self.b.modified.alarm.severity == AlarmSeverity.MINOR_ALARM
        assert self.b.modified.alarm.status == AlarmStatus.CONF_STATUS
        assert self.b.modified.alarm.message == \
            "part2.design.value = 'new_child' not 'init_child'"
        # Load the child again
        self.b_child.design.put_value("new_child")
        assert self.b.modified.value is True
        # And check that loading parent resets it
        self.b.design.put_value("init_parent")
        assert self.b.modified.value is False
        assert self.b_child.design.value == "init_child"
        # Put back
        self.b_child.design.put_value("new_child")
        assert self.b.modified.value is True
        # Do a configure, and check we get set back
        self.prepare_half_run()
        assert self.b_child.design.value == "init_child"
        assert self.b_child.modified.value is False
        assert self.b.modified.value is False

    def test_abort(self):
        self.b.abort()
        self.checkState(self.ss.ABORTED)

    def test_validate(self):
        line1 = LineGenerator('y', 'mm', 0, 2, 3)
        line2 = LineGenerator('x', 'mm', 0, 2, 2)
        compound = CompoundGenerator([line1, line2], [], [])
        actual = self.b.validate(generator=compound, axesToMove=['x'])
        assert actual["generator"].to_dict() == compound.to_dict()
        assert actual["axesToMove"] == ['x']

    def prepare_half_run(self, duration=0.01, exception=0):
        line1 = LineGenerator('y', 'mm', 0, 2, 3)
        line2 = LineGenerator('x', 'mm', 0, 2, 2)
        compound = CompoundGenerator([line1, line2], [], [], duration)
        self.b.configure(
            generator=compound, axesToMove=['x'], exceptionStep=exception)

    def test_configure_run(self):
        assert self.b.configure.writeable is True
        assert self.b.configure.takes.elements["generator"].writeable is True
        assert self.b.validate.takes.elements["generator"].writeable is True
        assert self.b.validate.returns.elements["generator"].writeable is False
        self.prepare_half_run()
        self.checkSteps(2, 0, 6)
        self.checkState(self.ss.ARMED)
        assert self.b.configure.writeable is False
        assert self.b.configure.takes.elements["generator"].writeable is True
        assert self.b.validate.takes.elements["generator"].writeable is True
        assert self.b.validate.returns.elements["generator"].writeable is False

        self.b.run()
        self.checkState(self.ss.ARMED)
        self.checkSteps(4, 2, 6)

        self.b.run()
        self.checkState(self.ss.ARMED)
        self.checkSteps(6, 4, 6)

        self.b.run()
        self.checkState(self.ss.READY)

    def test_abort(self):
        self.prepare_half_run()
        self.b.run()
        self.b.abort()
        self.checkState(self.ss.ABORTED)

    def test_pause_seek_resume(self):
        self.prepare_half_run()
        self.checkSteps(configured=2, completed=0, total=6)
        self.b.run()
        self.checkState(self.ss.ARMED)
        self.checkSteps(4, 2, 6)
        self.b.pause(completedSteps=1)
        self.checkState(self.ss.ARMED)
        self.checkSteps(2, 1, 6)
        self.b.run()
        self.checkSteps(4, 2, 6)
        self.b.completedSteps.put_value(5)
        self.checkSteps(6, 5, 6)
        self.b.run()
        self.checkState(self.ss.READY)

    def test_resume_in_run(self):
        self.prepare_half_run(duration=0.5)
        f = self.b.run_async()
        self.context.sleep(0.95)
        self.b.pause()
        self.checkState(self.ss.PAUSED)
        self.checkSteps(2, 1, 6)
        self.b.resume()
        # Parent should be running, child won't have got request yet
        then = time.time()
        self.checkState(self.ss.RUNNING, child=False)
        self.context.wait_all_futures(f, timeout=2)
        now = time.time()
        self.checkState(self.ss.ARMED)
        self.checkSteps(4, 2, 6)
        # This test fails on Travis sometimes, looks like the docker container
        # just gets starved
        # self.assertAlmostEqual(now - then, 0.5, delta=0.1)

    def test_run_exception(self):
        self.prepare_half_run(exception=1)
        with self.assertRaises(AssertionError):
            self.b.run()
        self.checkState(self.ss.FAULT)

    def test_run_stop(self):
        self.prepare_half_run(duration=0.1)
        f = self.b.run_async()
        self.context.sleep(0.1)
        self.b.abort()
        with self.assertRaises(AbortedError):
            f.result()
        self.checkState(self.ss.ABORTED)


# Make a ticker_block block to act as our child # Make an empty part for our parent # Make a RunnableChildPart to control the ticker_block # create a root block for the RunnableController block to reside in # start the process off # Save an initial setting for the child # x counter now at 31, child should be modified # x counter now at 2, child should be modified by us # x counter now at 0, child should be unmodified # Save an initial setting for child and parent # Change a value and save as a new child setting # Load the child again # And check that loading parent resets it # Put back # Do a configure, and check we get set back # Parent should be running, child won't have got request yet # This test fails on Travis sometimes, looks like the docker container # just gets starved # self.assertAlmostEqual(now - then, 0.5, delta=0.1)