from __future__ import annotations
import json
import os
import random
import sys
import uuid
from pathlib import Path
from typing import TYPE_CHECKING, Any, Generic
from .._types import PluginT, RawSettings
from ..flow.plugin_metadata import PluginMetadata
from ..query import Query
from ..settings import Settings
from ..utils import MISSING
from .filler import FillerObject
if TYPE_CHECKING:
from ..jsonrpc.responses import QueryResponse
from ..jsonrpc.results import Result
API_FILLER_TEXT = "FlowLauncherAPI is unavailable during testing. Consider passing the 'flow_api_client' arg into PluginTester to implement your own flow api client."
CHARACTERS = "qwertyuiopasdfghjklzxcvbnmQWERTYUIOPASDFGHJKLLZXCVBNM1234567890"
__all__ = ("PluginTester",)
[docs]
class PluginTester(Generic[PluginT]):
r"""This can be used to write tests for your plugins.
See the :doc:`testing` guide for more information on writing tests.
This class implements a generic for the :attr:`~flogin.testing.plugin_tester.PluginTester.plugin` attribute, which will be used for typechecking purposes.
Parameters
----------
plugin: :class:`~flogin.plugin.Plugin`
Your plugin
metadata: :class:`~flogin.flow.plugin_metadata.PluginMetadata` | dict[str, Any] | None
Your plugin's metadata. If ``None`` is passed, flogin will attempt to get the metadata from your ``plugin.json`` file. The :func:`PluginTester.create_plugin_metadata` and :func:`PluginTester.create_bogus_plugin_metadata` classmethods have been provided for creating :class:`~flogin.flow.plugin_metadata.PluginMetadata` objects.
flow_api_client: Optional[Any]
If not passed, flogin will use a filler class which will raise a runtime error whenever an attribute is accessed. If passed, you should be passing an instance of a class which will replace :class:`~flogin.flow.api.FlowLauncherAPI`, so make sure to implement the methods you need and handle them accordingly.
flow_version: Optional[:class:`str`]
This is an optional positional keyword that if set, will automatically set the enviroment variable ``FLOW_VERSION`` to the value. This is useful if your code uses the :attr:`~flogin.plugin.Plugin.flow_version` property.
.. versionadded: 1.1.0
flow_application_dir: Optional[:class:`str` | :class:`~pathlib.Path`]
This is an optional positional keyword that if set, will automatically set the enviroment variable ``FLOW_APPLICATION_DIRECTORY`` to the value. This is useful if your code uses the :attr:`~flogin.plugin.Plugin.flow_application_dir` property.
.. versionadded: 1.1.0
flow_program_dir: Optional[:class:`str` | :class:`~pathlib.Path`]
This is an optional positional keyword that if set, will automatically set the enviroment variable ``FLOW_PROGRAM_DIRECTORY`` to the value. This is useful if your code uses the :attr:`~flogin.plugin.Plugin.flow_program_dir` property.
.. versionadded: 1.1.0
Attributes
----------
plugin: :class:`~flogin.plugin.Plugin`
Your plugin
"""
plugin: PluginT
def __init__(
self,
plugin: PluginT,
*,
metadata: PluginMetadata | dict[str, Any] | None,
flow_api_client: Any = MISSING,
flow_version: str = MISSING,
flow_application_dir: Path | str = MISSING,
flow_program_dir: Path | str = MISSING,
) -> None:
self.plugin = plugin
if metadata is None:
if not os.path.exists("plugin.json"):
raise ValueError(
"plugin.json file can not be located, consider passing the metadata instead"
)
with open("plugin.json", "r") as f:
metadata = json.load(f)
assert metadata
if isinstance(metadata, dict):
metadata = PluginMetadata(metadata, self.plugin.api)
self.plugin._metadata = metadata
self.set_flow_api_client(flow_api_client)
if flow_version is not MISSING:
os.environ["FLOW_VERSION"] = flow_version
if flow_application_dir is not MISSING:
os.environ["FLOW_APPLICATION_DIRECTORY"] = str(flow_application_dir)
if flow_program_dir is not MISSING:
os.environ["FLOW_PROGRAM_DIRECTORY"] = str(flow_program_dir)
[docs]
def set_flow_api_client(self, flow_api_client: Any = MISSING) -> None:
r"""This sets the flow api client that the tests should use.
Parameters
----------
flow_api_client: Optional[Any]
If not passed, flogin will use a filler class which will raise a runtime error whenever an attribute is accessed. If passed, you should be passing an instance of a class which will replace :class:`~flogin.flow.api.FlowLauncherAPI`, so make sure to implement the methods you need and handle them accordingly.
"""
if flow_api_client is MISSING:
flow_api_client = FillerObject(API_FILLER_TEXT)
self.plugin.api = flow_api_client
self.plugin.metadata._flow_api = flow_api_client
[docs]
async def test_query(
self,
text: str,
*,
keyword: str = "*",
is_requery: bool = False,
settings: Settings | RawSettings | None = MISSING,
) -> QueryResponse:
r"""|coro|
This coroutine can be used to send your plugin a query, and get the response.
Parameters
----------
query: :class:`~flogin.query.Query`
The query object that should be passed to your search handlers.
settings: Optional[:class:`~flogin.settings.Settings` | dict[str, Any] | None]
This will represent the settings that flogin will use when executing your search handlers. If not passed, flogin will not use any settings. If ``None`` is passed, flogin will get the settings from the settings file (this is incompatible with :func:`PluginTester.create_bogus_plugin_metadata`). If a dict or :class:`~flogin.settings.Settings` object is passed, those are the settings that will be put in :attr:`~flogin.plugin.Plugin.settings` before executing your search handlers.
Returns
-------
:class:`~flogin.jsonrpc.responses.QueryResponse`
The query response object that would normally be sent to flow.
"""
if isinstance(settings, dict):
settings = Settings(settings)
if settings is MISSING:
settings = Settings({})
if isinstance(settings, Settings):
self.plugin.settings = settings
self.plugin._settings_are_populated = True
query = Query(
{
"rawQuery": f"{keyword} {text}",
"search": text,
"actionKeyword": keyword,
"isReQuery": is_requery,
},
self.plugin,
)
coro = self.plugin.process_search_handlers(query)
if coro is None:
raise RuntimeError("Query event handler not found")
return await coro # type: ignore
def __repr__(self):
return f"<PluginTester id={id(self)} {self.plugin=}>"