import string
from typing import cast
from unittest.mock import MagicMock

import pytest
import transformer.plugins as plug

from transformer.locust import locustfile, locust_taskset
from transformer.request import HttpMethod, Request, SplitResult
from transformer.scenario import Scenario
from transformer.task import Task, TIMEOUT
from transformer.plugins import plugin, Contract
from transformer.task import Task2
from transformer._version import __version__


class TestLocustfile:
    def test_it_renders_a_locustfile_template(self):
        a_name = "some_task"
        a_request = MagicMock(spec=Request)
        a_request.method = HttpMethod.GET
        a_request.url = MagicMock(spec=SplitResult)
        a_request.url.scheme = "some_scheme"
        a_request.url.hostname = "some_hostname"
        a_request.url.path = "some_path"
        a_request.url.geturl()
        a_request.url.geturl.return_value = "some_url"
        a_request.headers = {"a": "b"}
        a_request.name = None
        task = Task(a_name, a_request)
        scenario = Scenario(name="SomeScenario", children=[task], origin=None)
        scenario_group = Scenario(
            name="ScenarioGroup", children=[scenario], weight=2, origin=None
        )
        script = locustfile([scenario_group])
        expected = string.Template(
            f"""
# File automatically generated by Transformer v{__version__}:
# https://github.com/zalando-incubator/Transformer
import re
from distutils.version import LooseVersion
from locust import __version__
LOCUST_MAJOR_VERSION = LooseVersion(__version__).version[0]
if LOCUST_MAJOR_VERSION >= 1:
    from locust import HttpUser
    from locust import SequentialTaskSet
    from locust import TaskSet
    from locust import task
    HttpLocust = HttpUser
    TaskSequence = SequentialTaskSet
    def seq_task(_):
        return task
else:
    from locust import HttpLocust
    from locust import TaskSequence
    from locust import TaskSet
    from locust import seq_task
    from locust import task
class ScenarioGroup(TaskSet):
    @task(1)
    class SomeScenario(TaskSequence):
        @seq_task(1)
        def some_task(self):
            response = self.client.get(url='some_url', name='some_url', timeout=$TIMEOUT, allow_redirects=False, headers={{'a': 'b'}})
class LocustForScenarioGroup(HttpLocust):
    if LOCUST_MAJOR_VERSION >= 1:
        tasks = [ScenarioGroup]
    else:
        task_set = ScenarioGroup
    weight = 2
    min_wait = 0
    max_wait = 10
"""
        ).safe_substitute({"TIMEOUT": TIMEOUT})
        assert expected.strip() == script.strip()

    def test_it_renders_a_locustfile_template_with_plugin_change_task_name(self):
        @plugin(Contract.OnTask)
        def plugin_change_task_name(t: Task2) -> Task2:
            t.request.name = "changed_name"
            return t

        a_name = "some_task"
        a_request = MagicMock(spec=Request)
        a_request.method = HttpMethod.GET
        a_request.url = MagicMock(spec=SplitResult)
        a_request.url.scheme = "some_scheme"
        a_request.url.hostname = "some_hostname"
        a_request.url.path = "some_path"
        a_request.url.geturl()
        a_request.url.geturl.return_value = "some_url"
        a_request.headers = {"a": "b"}
        a_request.name = None
        task = plug.apply((plugin_change_task_name,), Task(a_name, a_request))
        scenario = Scenario(name="SomeScenario", children=[task], origin=None)
        scenario_group = Scenario(
            name="ScenarioGroup", children=[scenario], weight=2, origin=None
        )
        script = locustfile([scenario_group])
        expected = string.Template(
            f"""
# File automatically generated by Transformer v{__version__}:
# https://github.com/zalando-incubator/Transformer
import re
from distutils.version import LooseVersion
from locust import __version__
LOCUST_MAJOR_VERSION = LooseVersion(__version__).version[0]
if LOCUST_MAJOR_VERSION >= 1:
    from locust import HttpUser
    from locust import SequentialTaskSet
    from locust import TaskSet
    from locust import task
    HttpLocust = HttpUser
    TaskSequence = SequentialTaskSet
    def seq_task(_):
        return task
else:
    from locust import HttpLocust
    from locust import TaskSequence
    from locust import TaskSet
    from locust import seq_task
    from locust import task
class ScenarioGroup(TaskSet):
    @task(1)
    class SomeScenario(TaskSequence):
        @seq_task(1)
        def some_task(self):
            response = self.client.get(url='some_url', name='changed_name', timeout=$TIMEOUT, allow_redirects=False, headers={{'a': 'b'}})
class LocustForScenarioGroup(HttpLocust):
    if LOCUST_MAJOR_VERSION >= 1:
        tasks = [ScenarioGroup]
    else:
        task_set = ScenarioGroup
    weight = 2
    min_wait = 0
    max_wait = 10
"""
        ).safe_substitute({"TIMEOUT": TIMEOUT})
        assert expected.strip() == script.strip()


def test_generates_passed_global_code_blocks():
    def mock(name, blocks=None):
        m = MagicMock(spec=Scenario, children=[], global_code_blocks=blocks or {})
        # https://docs.python.org/3/library/unittest.mock.html#mock-names-and-the-name-attribute
        m.name = name
        return m

    sg1 = Scenario(
        "sg1",
        children=[mock("a", blocks={"b1": ["ab"]}), mock("b", blocks={"b2": ["cd"]})],
        origin=None,
    )
    sg2 = Scenario("sg2", children=[mock("c")], origin=None)
    sg3 = Scenario(
        "sg3",
        children=[mock("d", blocks={"b3": ["yz"], "b2": ["yyy", "zzz"]})],
        origin=None,
    )

    code = locustfile([sg1, sg2, sg3])
    assert code.endswith(
        "\n# b1\nab\n# b2\nyyy\nzzz\n# b3\nyz"
    ), "the latter b2 block should override the former"


def test_locust_taskset_raises_on_malformed_scenario():
    bad_child = cast(Scenario, 7)
    bad_scenario = Scenario(name="x", children=[bad_child], origin=None)
    with pytest.raises(TypeError, match=r"unexpected type .*\bchildren"):
        locust_taskset(bad_scenario)

import string
from typing import cast
from unittest.mock import MagicMock

import pytest
import transformer.plugins as plug

from transformer.locust import locustfile, locust_taskset
from transformer.request import HttpMethod, Request, SplitResult
from transformer.scenario import Scenario
from transformer.task import Task, TIMEOUT
from transformer.plugins import plugin, Contract
from transformer.task import Task2
from transformer._version import __version__


class TestLocustfile:
    def test_it_renders_a_locustfile_template(self):
        a_name = "some_task"
        a_request = MagicMock(spec=Request)
        a_request.method = HttpMethod.GET
        a_request.url = MagicMock(spec=SplitResult)
        a_request.url.scheme = "some_scheme"
        a_request.url.hostname = "some_hostname"
        a_request.url.path = "some_path"
        a_request.url.geturl()
        a_request.url.geturl.return_value = "some_url"
        a_request.headers = {"a": "b"}
        a_request.name = None
        task = Task(a_name, a_request)
        scenario = Scenario(name="SomeScenario", children=[task], origin=None)
        scenario_group = Scenario(
            name="ScenarioGroup", children=[scenario], weight=2, origin=None
        )
        script = locustfile([scenario_group])
        expected = string.Template(
            f"""
# File automatically generated by Transformer v{__version__}:
# https://github.com/zalando-incubator/Transformer
import re
from distutils.version import LooseVersion
from locust import __version__
LOCUST_MAJOR_VERSION = LooseVersion(__version__).version[0]
if LOCUST_MAJOR_VERSION >= 1:
    from locust import HttpUser
    from locust import SequentialTaskSet
    from locust import TaskSet
    from locust import task
    HttpLocust = HttpUser
    TaskSequence = SequentialTaskSet
    def seq_task(_):
        return task
else:
    from locust import HttpLocust
    from locust import TaskSequence
    from locust import TaskSet
    from locust import seq_task
    from locust import task
class ScenarioGroup(TaskSet):
    @task(1)
    class SomeScenario(TaskSequence):
        @seq_task(1)
        def some_task(self):
            response = self.client.get(url='some_url', name='some_url', timeout=$TIMEOUT, allow_redirects=False, headers={{'a': 'b'}})
class LocustForScenarioGroup(HttpLocust):
    if LOCUST_MAJOR_VERSION >= 1:
        tasks = [ScenarioGroup]
    else:
        task_set = ScenarioGroup
    weight = 2
    min_wait = 0
    max_wait = 10
"""
        ).safe_substitute({"TIMEOUT": TIMEOUT})
        assert expected.strip() == script.strip()

    def test_it_renders_a_locustfile_template_with_plugin_change_task_name(self):
        @plugin(Contract.OnTask)
        def plugin_change_task_name(t: Task2) -> Task2:
            t.request.name = "changed_name"
            return t

        a_name = "some_task"
        a_request = MagicMock(spec=Request)
        a_request.method = HttpMethod.GET
        a_request.url = MagicMock(spec=SplitResult)
        a_request.url.scheme = "some_scheme"
        a_request.url.hostname = "some_hostname"
        a_request.url.path = "some_path"
        a_request.url.geturl()
        a_request.url.geturl.return_value = "some_url"
        a_request.headers = {"a": "b"}
        a_request.name = None
        task = plug.apply((plugin_change_task_name,), Task(a_name, a_request))
        scenario = Scenario(name="SomeScenario", children=[task], origin=None)
        scenario_group = Scenario(
            name="ScenarioGroup", children=[scenario], weight=2, origin=None
        )
        script = locustfile([scenario_group])
        expected = string.Template(
            f"""
# File automatically generated by Transformer v{__version__}:
# https://github.com/zalando-incubator/Transformer
import re
from distutils.version import LooseVersion
from locust import __version__
LOCUST_MAJOR_VERSION = LooseVersion(__version__).version[0]
if LOCUST_MAJOR_VERSION >= 1:
    from locust import HttpUser
    from locust import SequentialTaskSet
    from locust import TaskSet
    from locust import task
    HttpLocust = HttpUser
    TaskSequence = SequentialTaskSet
    def seq_task(_):
        return task
else:
    from locust import HttpLocust
    from locust import TaskSequence
    from locust import TaskSet
    from locust import seq_task
    from locust import task
class ScenarioGroup(TaskSet):
    @task(1)
    class SomeScenario(TaskSequence):
        @seq_task(1)
        def some_task(self):
            response = self.client.get(url='some_url', name='changed_name', timeout=$TIMEOUT, allow_redirects=False, headers={{'a': 'b'}})
class LocustForScenarioGroup(HttpLocust):
    if LOCUST_MAJOR_VERSION >= 1:
        tasks = [ScenarioGroup]
    else:
        task_set = ScenarioGroup
    weight = 2
    min_wait = 0
    max_wait = 10
"""
        ).safe_substitute({"TIMEOUT": TIMEOUT})
        assert expected.strip() == script.strip()


def test_generates_passed_global_code_blocks():
    def mock(name, blocks=None):
        m = MagicMock(spec=Scenario, children=[], global_code_blocks=blocks or {})
        # https://docs.python.org/3/library/unittest.mock.html#mock-names-and-the-name-attribute
        m.name = name
        return m

    sg1 = Scenario(
        "sg1",
        children=[mock("a", blocks={"b1": ["ab"]}), mock("b", blocks={"b2": ["cd"]})],
        origin=None,
    )
    sg2 = Scenario("sg2", children=[mock("c")], origin=None)
    sg3 = Scenario(
        "sg3",
        children=[mock("d", blocks={"b3": ["yz"], "b2": ["yyy", "zzz"]})],
        origin=None,
    )

    code = locustfile([sg1, sg2, sg3])
    assert code.endswith(
        "\n# b1\nab\n# b2\nyyy\nzzz\n# b3\nyz"
    ), "the latter b2 block should override the former"


def test_locust_taskset_raises_on_malformed_scenario():
    bad_child = cast(Scenario, 7)
    bad_scenario = Scenario(name="x", children=[bad_child], origin=None)
    with pytest.raises(TypeError, match=r"unexpected type .*\bchildren"):
        locust_taskset(bad_scenario)

# File automatically generated by Transformer v{__version__}: # https://github.com/zalando-incubator/Transformer import re from distutils.version import LooseVersion from locust import __version__ LOCUST_MAJOR_VERSION = LooseVersion(__version__).version[0] if LOCUST_MAJOR_VERSION >= 1:     from locust import HttpUser     from locust import SequentialTaskSet     from locust import TaskSet     from locust import task     HttpLocust = HttpUser     TaskSequence = SequentialTaskSet     def seq_task(_):         return task else:     from locust import HttpLocust     from locust import TaskSequence     from locust import TaskSet     from locust import seq_task     from locust import task class ScenarioGroup(TaskSet):     @task(1)     class SomeScenario(TaskSequence):         @seq_task(1)         def some_task(self):             response = self.client.get(url='some_url', name='some_url', timeout=$TIMEOUT, allow_redirects=False, headers={{'a': 'b'}}) class LocustForScenarioGroup(HttpLocust):     if LOCUST_MAJOR_VERSION >= 1:         tasks = [ScenarioGroup]     else:         task_set = ScenarioGroup     weight = 2     min_wait = 0     max_wait = 10 # File automatically generated by Transformer v{__version__}: # https://github.com/zalando-incubator/Transformer import re from distutils.version import LooseVersion from locust import __version__ LOCUST_MAJOR_VERSION = LooseVersion(__version__).version[0] if LOCUST_MAJOR_VERSION >= 1:     from locust import HttpUser     from locust import SequentialTaskSet     from locust import TaskSet     from locust import task     HttpLocust = HttpUser     TaskSequence = SequentialTaskSet     def seq_task(_):         return task else:     from locust import HttpLocust     from locust import TaskSequence     from locust import TaskSet     from locust import seq_task     from locust import task class ScenarioGroup(TaskSet):     @task(1)     class SomeScenario(TaskSequence):         @seq_task(1)         def some_task(self):             response = self.client.get(url='some_url', name='changed_name', timeout=$TIMEOUT, allow_redirects=False, headers={{'a': 'b'}}) class LocustForScenarioGroup(HttpLocust):     if LOCUST_MAJOR_VERSION >= 1:         tasks = [ScenarioGroup]     else:         task_set = ScenarioGroup     weight = 2     min_wait = 0     max_wait = 10 # https://docs.python.org/3/library/unittest.mock.html#mock-names-and-the-name-attribute # b1\nab\n# b2\nyyy\nzzz\n# b3\nyz"