From dc6a8f2c7bfd72b6e3b680b6ba5a07326261a609 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Martin=20D=C3=B6rfelt?= Date: Tue, 27 Aug 2024 19:29:56 +0200 Subject: [PATCH] add support for the server API --- README.md | 23 +- joppy/client_api.py | 35 +- joppy/data_types.py | 175 ++++++++- joppy/server_api.py | 429 +++++++++++++++++++++++ joppy/tools.py | 18 + setup.cfg | 2 +- test/common.py | 69 ++++ test/setup_joplin.py | 41 ++- test/{test_api.py => test_client_api.py} | 100 ++---- test/test_server_api.py | 354 +++++++++++++++++++ 10 files changed, 1123 insertions(+), 123 deletions(-) create mode 100644 joppy/server_api.py create mode 100644 test/common.py rename test/{test_api.py => test_client_api.py} (93%) create mode 100644 test/test_server_api.py diff --git a/README.md b/README.md index 1e7634c..c29c40c 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # joppy -Python interface for the [Joplin data API](https://joplinapp.org/api/references/rest_api/). +Python interface for the [Joplin data API](https://joplinapp.org/api/references/rest_api/) (client) and the Joplin server API. [![build](https://github.com/marph91/joppy/actions/workflows/build.yml/badge.svg)](https://github.com/marph91/joppy/actions/workflows/build.yml) [![lint](https://github.com/marph91/joppy/actions/workflows/lint.yml/badge.svg)](https://github.com/marph91/joppy/actions/workflows/lint.yml) @@ -41,6 +41,8 @@ For details, consult the [implementation](joppy/api.py), [joplin documentation]( ## :bulb: Example snippets +### Client API + Start joplin and [get your API token](https://joplinapp.org/api/references/rest_api/#authorisation). Click to expand the examples.
@@ -218,7 +220,24 @@ for resource in api.get_all_resources():
-For more usage examples, check the example scripts or [tests](test/test_api.py). +For more usage examples, check the example scripts or [tests](test/test_client_api.py). + +### Server API + +The server API should work similarly to the client API in most cases. **Be aware that the server API is experimental and may break at any time. Make sure you have a backup and know how to restore it.** + +```python +from joppy.server_api import ServerApi + +# Create a new Api instance. +api = ServerApi(user="admin@localhost", password="admin", url="http://localhost:22300") + +# Add a notebook. +notebook_id = api.add_notebook(title="My first notebook") + +# Add a note in the previously created notebook. +note_id = api.add_note(title="My first note", body="With some content", parent_id=notebook_id) +``` ## :newspaper: Examples diff --git a/joppy/client_api.py b/joppy/client_api.py index 37bc291..6e85491 100644 --- a/joppy/client_api.py +++ b/joppy/client_api.py @@ -6,7 +6,6 @@ import time from typing import ( Any, - Callable, cast, Dict, List, @@ -18,6 +17,7 @@ import requests import joppy.data_types as dt +from joppy import tools # Use a global session object for better performance. @@ -33,12 +33,12 @@ ############################################################################## -# Base wrapper that manages the requests to the REST API. +# Base wrapper that manages the requests to the client REST API. ############################################################################## class ApiBase: - """Contains the basic requests of the REST API.""" + """Contains the basic requests of the client REST API.""" def __init__(self, token: str, url: str = "http://localhost:41184") -> None: self.url = url @@ -429,47 +429,32 @@ def delete_all_tags(self) -> None: assert tag.id is not None self.delete_tag(tag.id) - @staticmethod - def _unpaginate( - func: Callable[..., dt.DataList[dt.T]], **query: dt.JoplinTypes - ) -> List[dt.T]: - """Calls an Joplin endpoint until it's response doesn't contain more data.""" - response = func(**query) - items = response.items - page = 1 # pages are one based - while response.has_more: - page += 1 - query["page"] = page - response = func(**query) - items.extend(response.items) - return items - def get_all_events(self, **query: dt.JoplinTypes) -> List[dt.EventData]: """Get all events, unpaginated.""" - return self._unpaginate(self.get_events, **query) + return tools._unpaginate(self.get_events, **query) def get_all_notes(self, **query: dt.JoplinTypes) -> List[dt.NoteData]: """Get all notes, unpaginated.""" - return self._unpaginate(self.get_notes, **query) + return tools._unpaginate(self.get_notes, **query) def get_all_notebooks(self, **query: dt.JoplinTypes) -> List[dt.NotebookData]: """Get all notebooks, unpaginated.""" - return self._unpaginate(self.get_notebooks, **query) + return tools._unpaginate(self.get_notebooks, **query) def get_all_resources(self, **query: dt.JoplinTypes) -> List[dt.ResourceData]: """Get all resources, unpaginated.""" - return self._unpaginate(self.get_resources, **query) + return tools._unpaginate(self.get_resources, **query) def get_all_revisions(self, **query: dt.JoplinTypes) -> List[dt.RevisionData]: """Get all revisions, unpaginated.""" - return self._unpaginate(self.get_revisions, **query) + return tools._unpaginate(self.get_revisions, **query) def get_all_tags(self, **query: dt.JoplinTypes) -> List[dt.TagData]: """Get all tags, unpaginated.""" - return self._unpaginate(self.get_tags, **query) + return tools._unpaginate(self.get_tags, **query) def search_all( self, **query: dt.JoplinTypes ) -> List[Union[dt.NoteData, dt.NotebookData, dt.ResourceData, dt.TagData]]: """Issue a search and get all results, unpaginated.""" - return self._unpaginate(self.search, **query) # type: ignore + return tools._unpaginate(self.search, **query) # type: ignore diff --git a/joppy/data_types.py b/joppy/data_types.py index 4eb8bc6..1d891dc 100644 --- a/joppy/data_types.py +++ b/joppy/data_types.py @@ -12,6 +12,7 @@ Set, TypeVar, ) +import uuid # Datatypes used by the Joplin API. Needed for arbitrary kwargs. @@ -20,14 +21,14 @@ JoplinKwargs = MutableMapping[str, JoplinTypes] -class EventChangeType(enum.Enum): +class EventChangeType(enum.IntEnum): # https://joplinapp.org/api/references/rest_api/#properties-4 CREATED = 1 UPDATED = 2 DELETED = 3 -class ItemType(enum.Enum): +class ItemType(enum.IntEnum): # https://joplinapp.org/api/references/rest_api/#item-type-ids NOTE = 1 FOLDER = 2 @@ -47,7 +48,7 @@ class ItemType(enum.Enum): COMMAND = 16 -class MarkupLanguage(enum.Enum): +class MarkupLanguage(enum.IntEnum): # https://discourse.joplinapp.org/t/api-body-vs-body-html/11697/4 MARKDOWN = 1 HTML = 2 @@ -93,10 +94,18 @@ def __post_init__(self) -> None: "todo_due", "todo_completed", ): - casted_value = ( - None if value == 0 else datetime.fromtimestamp(value / 1000.0) - ) - setattr(self, field_.name, casted_value) + try: + value_int = int(value) + casted_value = ( + None + if value_int == 0 + else datetime.fromtimestamp(value_int / 1000.0) + ) + setattr(self, field_.name, casted_value) + except ValueError: + # TODO: This is not spec conform. + casted_value = datetime.strptime(value, "%Y-%m-%dT%H:%M:%S.%fZ") + setattr(self, field_.name, casted_value) elif field_.name in ( "is_conflict", "is_todo", @@ -104,23 +113,25 @@ def __post_init__(self) -> None: "is_shared", "encryption_blob_encrypted", ): - setattr(self, field_.name, bool(value)) + setattr(self, field_.name, bool(int(value))) elif field_.name == "latitude": - if not (-90 <= value <= 90): + setattr(self, field_.name, float(value)) + if not (-90 <= float(value) <= 90): raise ValueError("Invalid latitude:", value) elif field_.name == "longitude": - if not (-180 <= value <= 180): + setattr(self, field_.name, float(value)) + if not (-180 <= float(value) <= 180): raise ValueError("Invalid longitude:", value) elif field_.name == "markup_language": - setattr(self, field_.name, MarkupLanguage(value)) + setattr(self, field_.name, MarkupLanguage(int(value))) # elif field_.name == "order": # elif field_.name == "crop_rect": # elif field_.name == "icon": # elif field_.name == "filename": # "file_extension" elif field_.name in ("item_type", "type_"): - setattr(self, field_.name, ItemType(value)) + setattr(self, field_.name, ItemType(int(value))) elif field_.name == "type": - setattr(self, field_.name, EventChangeType(value)) + setattr(self, field_.name, EventChangeType(int(value))) def assigned_fields(self) -> Set[str]: # Exclude "type_" for convenience. @@ -139,7 +150,7 @@ def fields(cls) -> Set[str]: def default_fields() -> Set[str]: return {"id", "parent_id", "title"} - def __repr__(self) -> str: + def __str__(self) -> str: # show only fields with values not_none_fields = ", ".join( f"{field.name}={getattr(self, field.name)}" @@ -188,6 +199,44 @@ class NoteData(BaseData): image_data_url: Optional[str] = None crop_rect: Optional[str] = None + def serialize(self) -> str: + lines = [] + if self.title is not None: + lines.extend([self.title, ""]) + if self.body is not None: + lines.extend([self.body, ""]) + for field_ in fields(self): + match field_.name: + case "id": + # ID is always required + if self.id is None: + self.id = uuid.uuid4().hex + lines.append(f"{field_.name}: {self.id}") + case "markup_language": + # required to get an editable note + if self.markup_language is None: + self.markup_language = MarkupLanguage.MARKDOWN + lines.append(f"{field_.name}: {self.markup_language}") + case "source_application": + if self.source_application is None: + self.source_application = "joppy" + lines.append(f"{field_.name}: {self.source_application}") + case "title" | "body": + pass # handled before + case "type_": + self.item_type = ItemType.NOTE + lines.append(f"{field_.name}: {self.item_type}") + case "updated_time": + # required, even if empty + value_raw = getattr(self, field_.name) + value = "" if value_raw is None else value_raw + lines.append(f"{field_.name}: {value}") + case _: + value_raw = getattr(self, field_.name) + if value_raw is not None: + lines.append(f"{field_.name}: {value_raw}") + return "\n".join(lines) + @dataclass class NotebookData(BaseData): @@ -209,6 +258,33 @@ class NotebookData(BaseData): user_data: Optional[str] = None deleted_time: Optional[datetime] = None + def serialize(self) -> str: + lines = [] + if self.title is not None: + lines.extend([self.title, ""]) + for field_ in fields(self): + match field_.name: + case "id": + # ID is always required + if self.id is None: + self.id = uuid.uuid4().hex + lines.append(f"{field_.name}: {self.id}") + case "title": + pass # handled before + case "type_": + self.item_type = ItemType.FOLDER + lines.append(f"{field_.name}: {self.item_type}") + case "updated_time": + # required, even if empty + value_raw = getattr(self, field_.name) + value = "" if value_raw is None else value_raw + lines.append(f"{field_.name}: {value}") + case _: + value_raw = getattr(self, field_.name) + if value_raw is not None: + lines.append(f"{field_.name}: {value_raw}") + return "\n".join(lines) + @dataclass class ResourceData(BaseData): @@ -280,6 +356,72 @@ class TagData(BaseData): parent_id: Optional[str] = None user_data: Optional[str] = None + def serialize(self) -> str: + lines = [] + if self.title is not None: + lines.extend([self.title, ""]) + for field_ in fields(self): + match field_.name: + case "id": + # ID is always required + if self.id is None: + self.id = uuid.uuid4().hex + lines.append(f"{field_.name}: {self.id}") + case "title": + pass # handled before + case "type_": + self.item_type = ItemType.TAG + lines.append(f"{field_.name}: {self.item_type}") + case "updated_time": + # required, even if empty + value_raw = getattr(self, field_.name) + value = "" if value_raw is None else value_raw + lines.append(f"{field_.name}: {value}") + case _: + value_raw = getattr(self, field_.name) + if value_raw is not None: + lines.append(f"{field_.name}: {value_raw}") + return "\n".join(lines) + + +@dataclass +class NoteTagData(BaseData): + """Links a tag to a note.""" + + id: Optional[str] = None + note_id: Optional[str] = None + tag_id: Optional[str] = None + created_time: Optional[datetime] = None + updated_time: Optional[datetime] = None + user_created_time: Optional[datetime] = None + user_updated_time: Optional[datetime] = None + encryption_cipher_text: Optional[str] = None + encryption_applied: Optional[bool] = None + is_shared: Optional[bool] = None + + def serialize(self) -> str: + lines = [] + for field_ in fields(self): + match field_.name: + case "id": + # ID is always required + if self.id is None: + self.id = uuid.uuid4().hex + lines.append(f"{field_.name}: {self.id}") + case "type_": + self.item_type = ItemType.NOTE_TAG + lines.append(f"{field_.name}: {self.item_type}") + case "updated_time": + # required, even if empty + value_raw = getattr(self, field_.name) + value = "" if value_raw is None else value_raw + lines.append(f"{field_.name}: {value}") + case _: + value_raw = getattr(self, field_.name) + if value_raw is not None: + lines.append(f"{field_.name}: {value_raw}") + return "\n".join(lines) + @dataclass class EventData(BaseData): @@ -304,6 +446,11 @@ def default_fields() -> Set[str]: return {"id", "item_type", "item_id", "type", "created_time"} +AnyData = Union[ + EventData, NoteData, NotebookData, NoteTagData, ResourceData, RevisionData, TagData +] + + T = TypeVar( "T", EventData, NoteData, NotebookData, ResourceData, RevisionData, TagData, str ) diff --git a/joppy/server_api.py b/joppy/server_api.py new file mode 100644 index 0000000..b6b3815 --- /dev/null +++ b/joppy/server_api.py @@ -0,0 +1,429 @@ +import logging +import pathlib +from typing import Any, cast, Dict, List, Optional + +import requests + +import joppy.data_types as dt +from joppy import tools + + +# Use a global session object for better performance. +# Define it globally to avoid "ResourceWarning". +SESSION = requests.Session() + +# Don't spam the log. See: https://stackoverflow.com/a/11029841/7410886. +logging.getLogger("requests").setLevel(logging.WARNING) +logging.getLogger("urllib3").setLevel(logging.WARNING) + +LOGGER = logging.getLogger("joppy") + + +def deserialize(body: str) -> Optional[dt.AnyData]: + """Deserialize server data from string to a known data type.""" + # https://github.com/laurent22/joplin/blob/b617a846964ea49be2ffefd31439e911ad84ed8c/packages/lib/models/BaseItem.ts#L549-L596 + body_splitted = body.split("\n\n") + + def extract_metadata(serialized_metadata: str) -> Dict[Any, Any]: + metadata = {} + for line in serialized_metadata.split("\n"): + key, value = line.split(": ", 1) + if not value: + continue + metadata[key] = value + return metadata + + match len(body_splitted): + case 1: + # metadata only + title = None + note_body = None + metadata = extract_metadata(body_splitted[0]) + case 2: + # title + metadata + title = body_splitted[0] + note_body = None + metadata = extract_metadata(body_splitted[1]) + case 3: + # title + body + metadata + title = body_splitted[0] + note_body = body_splitted[1] + metadata = extract_metadata(body_splitted[2]) + case _: + print("TODO: ", body_splitted) + if title is not None: + metadata["title"] = title + if note_body is not None: + metadata["body"] = note_body + + match dt.ItemType(int(metadata["type_"])): + case dt.ItemType.NOTE: + return dt.NoteData(**metadata) + case dt.ItemType.FOLDER: + return dt.NotebookData(**metadata) + case dt.ItemType.TAG: + return dt.TagData(**metadata) + case dt.ItemType.NOTE_TAG: + return dt.NoteTagData(**metadata) + case dt.ItemType.REVISION: + # ignore revisions for now + pass + case _: + print("TODO: ", dt.ItemType(int(metadata["type_"]))) + return None + + +def add_suffix(string: str, suffix: str = ".md") -> str: + """Add a suffix.""" + return str(pathlib.Path(string).with_suffix(suffix)) + + +def remove_suffix(string: str) -> str: + """Remove the suffix.""" + return str(pathlib.Path(string).with_suffix("")) + + +############################################################################## +# Base wrapper that manages the requests to the client REST API. +############################################################################## + + +class ApiBase: + """Contains the basic requests of the server REST API.""" + + def __init__( + self, + user: str = "admin@localhost", + password: str = "admin", + url: str = "http://localhost:22300", + encryption_key: Optional[str] = None, + ) -> None: + self.url = url + self.encryption_key = encryption_key # TODO + + # cookie is saved in session and used for the next requests + self.post("/login", data={"email": user, "password": password}) + + def _request( + self, + method: str, + path: str, + query: Optional[dt.JoplinKwargs] = None, + data: Any = None, + files: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, Any]] = None, + ) -> requests.models.Response: + LOGGER.debug(f"API: {method} request: path={path}, query={query}, data={data}") + if query is None: + query = {} + query_str = "&".join([f"{key}={val}" for key, val in query.items()]) + + try: + response: requests.models.Response = getattr(SESSION, method)( + f"{self.url}{path}?{query_str}", + data=data, + files=files, + headers=headers, + ) + LOGGER.debug(f"API: response {response.text}") + response.raise_for_status() + except requests.exceptions.HTTPError as err: + err.args = err.args + (response.text,) + raise + return response + + def delete( + self, path: str, query: Optional[dt.JoplinKwargs] = None + ) -> requests.models.Response: + """Convenience method to issue a delete request.""" + return self._request("delete", path, query=query) + + def get( + self, path: str, query: Optional[dt.JoplinKwargs] = None + ) -> requests.models.Response: + """Convenience method to issue a get request.""" + return self._request("get", path, query=query) + + def post( + self, + path: str, + data: Optional[dt.JoplinKwargs] = None, + files: Optional[Dict[str, Any]] = None, + ) -> requests.models.Response: + """Convenience method to issue a post request.""" + return self._request("post", path, data=data, files=files) + + def put(self, path: str, data: str) -> requests.models.Response: + """Convenience method to issue a put request.""" + return self._request( + "put", path, data=data, headers={"Content-Type": "application/octet-stream"} + ) + + +############################################################################## +# Specific classes +############################################################################## + + +# TODO +class Event(ApiBase): + pass + + +class Note(ApiBase): + def add_note(self, **data: Any) -> str: + """Add a note.""" + note_data = dt.NoteData(**data) + request_data = note_data.serialize() + assert note_data.id is not None + self.put( + f"/api/items/root:/{add_suffix(note_data.id)}:/content", data=request_data + ) + return note_data.id + + def delete_note(self, id_: str) -> None: + """Delete a note.""" + self.delete(f"/api/items/{add_suffix(id_)}") + + def get_note(self, id_: str) -> dt.NoteData: + response = self.get(f"/api/items/{add_suffix(id_)}/content") + return cast(dt.NoteData, deserialize(response.text)) + + def get_notes(self) -> dt.DataList[dt.NoteData]: + response = self.get("/api/items/root:/:/children").json() + # TODO: Is this the best practice? + notes = [] + for item in response["items"]: + if item["name"].endswith(".md"): + item_complete = self.get_note(remove_suffix(item["name"])) + if isinstance(item_complete, dt.NoteData): + notes.append(item_complete) + return dt.DataList(response["has_more"], response["cursor"], notes) + + def modify_note(self, id_: str, **data: Any) -> None: + """Modify a note.""" + # TODO: Without fetching the orginal note, this would be replacing, + # not modifying. + id_server = add_suffix(id_) + note_data = self.get_note(id_server) + for key, value in data.items(): + setattr(note_data, key, value) + request_data = note_data.serialize() + self.put(f"/api/items/root:/{id_server}:/content", data=request_data) + + +class Notebook(ApiBase): + def add_notebook(self, **data: Any) -> str: + """Add a notebook.""" + notebook_data = dt.NotebookData(**data) + request_data = notebook_data.serialize() + assert notebook_data.id is not None + self.put( + f"/api/items/root:/{add_suffix(notebook_data.id)}:/content", + data=request_data, + ) + return notebook_data.id + + def delete_notebook(self, id_: str) -> None: + """Delete a notebook.""" + self.delete(f"/api/items/{add_suffix(id_)}") + + def get_notebook(self, id_: str) -> dt.NotebookData: + response = self.get(f"/api/items/{add_suffix(id_)}/content") + return cast(dt.NotebookData, deserialize(response.text)) + + def get_notebooks(self) -> dt.DataList[dt.NotebookData]: + response = self.get("/api/items/root:/:/children").json() + # TODO: Is this the best practice? + notebooks = [] + for item in response["items"]: + if item["name"].endswith(".md"): + item_complete = self.get_notebook(remove_suffix(item["name"])) + if isinstance(item_complete, dt.NotebookData): + notebooks.append(item_complete) + return dt.DataList(response["has_more"], response["cursor"], notebooks) + + def modify_notebook(self, id_: str, **data: Any) -> None: + """Modify a notebook.""" + # TODO: Without fetching the orginal notebook, this would be replacing, + # not modifying. + id_server = add_suffix(id_) + notebook_data = self.get_notebook(id_server) + for key, value in data.items(): + setattr(notebook_data, key, value) + request_data = notebook_data.serialize() + self.put(f"/api/items/root:/{id_server}:/content", data=request_data) + + +class Ping(ApiBase): + def ping(self) -> requests.models.Response: + """Ping the API.""" + return self.get("/api/ping") + + +# TODO +class Resource(ApiBase): + pass + + +class Revision(ApiBase): + def delete_revision(self, id_: str) -> None: + """Delete a revision.""" + self.delete(f"/api/items/{add_suffix(id_)}") + + def get_revision(self, id_: str, **query: Any) -> dt.RevisionData: + """Get the revision with the given ID.""" + response = self.get(f"/api/items/{add_suffix(id_)}/content") + return cast(dt.RevisionData, deserialize(response.text)) + + def get_revisions(self, **query: Any) -> dt.DataList[dt.RevisionData]: + response = self.get("/api/items/root:/:/children").json() + # TODO: Is this the best practice? + revisions = [] + for item in response["items"]: + if item["name"].endswith(".md"): + item_complete = self.get_revision(remove_suffix(item["name"])) + if isinstance(item_complete, dt.RevisionData): + revisions.append(item_complete) + return dt.DataList(response["has_more"], response["cursor"], revisions) + + +# TODO: Is there a search functionality? +class Search(ApiBase): + pass + + +class Tag(ApiBase): + def add_tag(self, **data: Any) -> str: + """Add a tag.""" + tag_data = dt.TagData(**data) + request_data = tag_data.serialize() + assert tag_data.id is not None + self.put( + f"/api/items/root:/{add_suffix(tag_data.id)}:/content", data=request_data + ) + return tag_data.id + + def delete_tag(self, id_: str) -> None: + """Delete a tag.""" + self.delete(f"/api/items/{add_suffix(id_)}") + + def get_tag(self, id_: str) -> dt.TagData: + """Get the tag with the given ID.""" + response = self.get(f"/api/items/{add_suffix(id_)}/content") + return cast(dt.TagData, deserialize(response.text)) + + def get_tags(self) -> dt.DataList[dt.TagData]: + """ + Get tags, paginated. If a note is given, return the corresponding tags. + To get all tags (unpaginated), use "get_all_tags()". + """ + response = self.get("/api/items/root:/:/children").json() + # TODO: Is this the best practice? + tags = [] + for item in response["items"]: + if item["name"].endswith(".md"): + item_complete = self.get_tag(remove_suffix(item["name"])) + if isinstance(item_complete, dt.TagData): + tags.append(item_complete) + return dt.DataList(response["has_more"], response["cursor"], tags) + + def modify_tag(self, id_: str, **data: Any) -> None: + """Modify a tag.""" + # TODO: Without fetching the orginal tag, this would be replacing, + # not modifying. + id_server = add_suffix(id_) + tag_data = self.get_tag(id_server) + for key, value in data.items(): + setattr(tag_data, key, value) + request_data = tag_data.serialize() + self.put(f"/api/items/root:/{id_server}:/content", data=request_data) + + +class ServerApi(Event, Note, Notebook, Ping, Resource, Revision, Search, Tag): + """ + Collects all basic API functions and contains a few more useful methods. + This should be the only class accessed from the users. + """ + + def add_tag_to_note(self, tag_id: str, note_id: str) -> str: + """Add a tag to a given note.""" + note_tag_data = dt.NoteTagData(tag_id=tag_id, note_id=note_id) + request_data = note_tag_data.serialize() + assert note_tag_data.id is not None + self.put( + f"/api/items/root:/{add_suffix(note_tag_data.id)}:/content", + data=request_data, + ) + return note_tag_data.id + + # TODO + # def add_resource_to_note(self, resource_id: str, note_id: str) -> None: + # """Add a resource to a given note.""" + # note = self.get_note(id_=note_id, fields="body") + # resource = self.get_resource(id_=resource_id, fields="title,mime") + # # TODO: Use "assertIsNotNone()" when + # # https://github.com/python/mypy/issues/5528 is resolved. + # assert resource.mime is not None + # image_prefix = "!" if resource.mime.startswith("image/") else "" + # body_with_attachment = ( + # f"{note.body}\n{image_prefix}[{resource.title}](:/{resource_id})" + # ) + # self.modify_note(note_id, body=body_with_attachment) + + def delete_all_notes(self) -> None: + """Delete all notes permanently.""" + for note in self.get_all_notes(): + assert note.id is not None + self.delete_note(note.id) + + def delete_all_notebooks(self) -> None: + """Delete all notebooks permanently.""" + for notebook in self.get_all_notebooks(): + assert notebook.id is not None + self.delete_notebook(notebook.id) + + def delete_all_resources(self) -> None: + """Delete all resources.""" + pass # TODO + # for resource in self.get_all_resources(): + # assert resource.id is not None + # self.delete_resource(resource.id) + + def delete_all_revisions(self) -> None: + """Delete all revisions.""" + for revision in self.get_all_revisions(): + assert revision.id is not None + self.delete_revision(revision.id) + + def delete_all_tags(self) -> None: + """Delete all tags.""" + for tag in self.get_all_tags(): + assert tag.id is not None + self.delete_tag(tag.id) + + # TODO + # def get_all_events(self, **query: Any) -> List[dt.EventData]: + # """Get all events, unpaginated.""" + # return self._unpaginate(self.get_events, **query) + + def get_all_notes(self, **query: Any) -> List[dt.NoteData]: + """Get all notes, unpaginated.""" + return tools._unpaginate(self.get_notes, **query) + + def get_all_notebooks(self, **query: Any) -> List[dt.NotebookData]: + """Get all notebooks, unpaginated.""" + return tools._unpaginate(self.get_notebooks, **query) + + def get_all_resources(self, **query: Any) -> List[dt.ResourceData]: + """Get all resources, unpaginated.""" + return [] # TODO + # return tools._unpaginate(self.get_resources, **query) + + def get_all_revisions(self, **query: Any) -> List[dt.RevisionData]: + """Get all revisions, unpaginated.""" + return tools._unpaginate(self.get_revisions, **query) + + def get_all_tags(self, **query: Any) -> List[dt.TagData]: + """Get all tags, unpaginated.""" + return tools._unpaginate(self.get_tags, **query) diff --git a/joppy/tools.py b/joppy/tools.py index 06ec840..b6e29a1 100644 --- a/joppy/tools.py +++ b/joppy/tools.py @@ -1,9 +1,27 @@ """Helper functions for the API.""" import base64 +from typing import Callable, List + +import joppy.data_types as dt def encode_base64(filepath: str) -> str: """Encode an arbitrary file to base64.""" with open(filepath, "rb") as img_file: return base64.b64encode(img_file.read()).decode("utf-8") + + +def _unpaginate( + func: Callable[..., dt.DataList[dt.T]], **query: dt.JoplinTypes +) -> List[dt.T]: + """Calls an Joplin endpoint until it's response doesn't contain more data.""" + response = func(**query) + items = response.items + page = 1 # pages are one based + while response.has_more: + page += 1 + query["page"] = page + response = func(**query) + items.extend(response.items) + return items diff --git a/setup.cfg b/setup.cfg index bc0ed34..aae5384 100644 --- a/setup.cfg +++ b/setup.cfg @@ -21,7 +21,7 @@ ignore_missing_imports = True [mypy-setuptools.*] ignore_missing_imports = True # Signature types of tests don't matter. -[mypy-test.test_api.*] +[mypy-test.*] disallow_untyped_defs = False [mypy-xvfbwrapper.*] ignore_missing_imports = True \ No newline at end of file diff --git a/test/common.py b/test/common.py new file mode 100644 index 0000000..19d2126 --- /dev/null +++ b/test/common.py @@ -0,0 +1,69 @@ +import itertools +import logging +import os +import random +import string +from typing import Any, Iterable, Tuple +import unittest + + +os.makedirs("test_output", exist_ok=True) +logging.basicConfig( + filename="test_output/test.log", + filemode="w", + format="%(asctime)s [%(levelname)s]: %(message)s", + level=logging.DEBUG, +) +LOGGER = logging.getLogger("joppy") + +SLOW_TESTS = bool(os.getenv("SLOW_TESTS", "")) + + +class Base(unittest.TestCase): + api: Any + + def setUp(self) -> None: + super().setUp() + + LOGGER.debug("Test: %s", self.id()) + + self.api.delete_all_notebooks() + self.api.delete_all_notes() + self.api.delete_all_resources() + self.api.delete_all_tags() + # Delete revisions last to cover all previous deletions. + self.api.delete_all_revisions() + + @staticmethod + def get_random_id() -> str: + """Return a random, valid ID.""" + # https://stackoverflow.com/a/2782859/7410886 + return f"{random.randrange(16**32):032x}" + + @staticmethod + def get_random_string(length: int = 8, exclude: str = "") -> str: + """Return a random string.""" + characters = string.printable + for character in exclude: + characters = characters.replace(character, "") + random_string = "".join(random.choice(characters) for _ in range(length)) + LOGGER.debug("Test: random string: %s", random_string) + return random_string + + @staticmethod + def get_combinations( + iterable: Iterable[str], max_combinations: int = 100 + ) -> Iterable[Tuple[str, ...]]: + """Get some combinations of an iterable.""" + # https://stackoverflow.com/a/10465588 + # TODO: Randomize fully. For now the combinations are sorted by length. + list_ = list(iterable) + lengths = list(range(1, len(list_) + 1)) + random.shuffle(lengths) + combinations = itertools.chain.from_iterable( + itertools.combinations(list_, r) + for r in lengths + # shuffle each iteration + if random.shuffle(list_) is None + ) + return itertools.islice(combinations, max_combinations) diff --git a/test/setup_joplin.py b/test/setup_joplin.py index a13c6b3..8916819 100644 --- a/test/setup_joplin.py +++ b/test/setup_joplin.py @@ -11,7 +11,7 @@ from xvfbwrapper import Xvfb -def download_joplin(destination: str) -> None: +def download_joplin_client(destination: str) -> None: """Download the latest joplin desktop app release if not already done.""" if not os.path.exists(destination): # obtain the version string @@ -66,8 +66,8 @@ def wait_for(func: Callable[..., Any], interval: float = 0.5, timeout: int = 5) raise TimeoutError(f"Function didn't return a valid value {result}.") -class JoplinApp: - """Represents a joplin application.""" +class JoplinClient: + """Represents a joplin client application.""" def __init__(self, app_path: str, profile: str): self.xvfb = xvfb = Xvfb() @@ -108,3 +108,38 @@ def stop(self) -> None: self.xvfb.stop() self.joplin_process.terminate() self.joplin_process.wait() + + +class JoplinServer: + """Represents a joplin server.""" + + def __init__(self) -> None: + # Wait until the API is available. + def api_available() -> Optional[bool]: + try: + response = requests.get("http://localhost:22300/api/ping", timeout=5) + if response.status_code == 200: + return True + except Exception: + pass + return None + + # check if server is running already + if api_available() is not None: + self.joplin_process = None + return + + # TODO: check if docker is installed and available + self.joplin_process = subprocess.Popen( + ["docker", "run", "-p", "22300:22300", "joplin/server:latest"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + wait_for(api_available, timeout=20) + + def stop(self) -> None: + """Stop the joplin server.""" + if self.joplin_process is not None: + self.joplin_process.terminate() + self.joplin_process.wait() diff --git a/test/test_api.py b/test/test_client_api.py similarity index 93% rename from test/test_api.py rename to test/test_client_api.py index ecc96fe..24ae6a4 100644 --- a/test/test_api.py +++ b/test/test_client_api.py @@ -1,15 +1,13 @@ -"""Tests for the joplin python API.""" +"""Tests for the Joplin client python API.""" from datetime import datetime -import itertools -import logging import mimetypes import os import random import re import string import tempfile -from typing import Any, Iterable, Mapping, Tuple +from typing import Any, Mapping import unittest import requests @@ -18,20 +16,9 @@ from joppy import tools from joppy.client_api import ClientApi import joppy.data_types as dt -from . import setup_joplin +from . import common, setup_joplin -os.makedirs("test_output", exist_ok=True) -logging.basicConfig( - filename="test_output/test.log", - filemode="w", - format="%(asctime)s [%(levelname)s]: %(message)s", - level=logging.DEBUG, -) -LOGGER = logging.getLogger("joppy") - - -SLOW_TESTS = bool(os.getenv("SLOW_TESTS", "")) PROFILE = os.path.join(os.getcwd(), "test_profile") API_TOKEN = "" # Don't use the API token from env to avoid data loss. APP = None @@ -57,8 +44,8 @@ def setUpModule(): # pylint: disable=invalid-name global API_TOKEN, APP if not API_TOKEN: app_path = "./joplin.AppImage" - setup_joplin.download_joplin(app_path) - APP = setup_joplin.JoplinApp(app_path, PROFILE) + setup_joplin.download_joplin_client(app_path) + APP = setup_joplin.JoplinClient(app_path, PROFILE) API_TOKEN = APP.api_token @@ -67,56 +54,13 @@ def tearDownModule(): # pylint: disable=invalid-name APP.stop() -class TestBase(unittest.TestCase): +class ClientBase(common.Base): def setUp(self): - super().setUp() - - LOGGER.debug("Test: %s", self.id()) - self.api = ClientApi(token=API_TOKEN) - # Notes get deleted automatically. - self.api.delete_all_notebooks() - self.api.delete_all_resources() - self.api.delete_all_tags() - # Delete revisions last to cover all previous deletions. - self.api.delete_all_revisions() - - @staticmethod - def get_random_id() -> str: - """Return a random, valid ID.""" - # https://stackoverflow.com/a/2782859/7410886 - return f"{random.randrange(16**32):032x}" - - @staticmethod - def get_random_string(length: int = 8, exclude: str = "") -> str: - """Return a random string.""" - characters = string.printable - for character in exclude: - characters = characters.replace(character, "") - random_string = "".join(random.choice(characters) for _ in range(length)) - LOGGER.debug("Test: random string: %s", random_string) - return random_string - - @staticmethod - def get_combinations( - iterable: Iterable[str], max_combinations: int = 100 - ) -> Iterable[Tuple[str, ...]]: - """Get some combinations of an iterable.""" - # https://stackoverflow.com/a/10465588 - # TODO: Randomize fully. For now the combinations are sorted by length. - list_ = list(iterable) - lengths = list(range(1, len(list_) + 1)) - random.shuffle(lengths) - combinations = itertools.chain.from_iterable( - itertools.combinations(list_, r) - for r in lengths - # shuffle each iteration - if random.shuffle(list_) is None - ) - return itertools.islice(combinations, max_combinations) + super().setUp() -class Event(TestBase): +class Event(ClientBase): def generate_event(self) -> None: """Generate an event and wait until it's available.""" events = self.api.get_events() @@ -189,7 +133,7 @@ def test_get_events_valid_properties(self): self.assertEqual(event.assigned_fields(), set(properties)) -class Note(TestBase): +class Note(ClientBase): def test_add(self): """Add a note to an existing notebook.""" parent_id = self.api.add_notebook() @@ -316,7 +260,7 @@ def test_move(self): self.assertEqual(note.title, original_title) -class Notebook(TestBase): +class Notebook(ClientBase): def test_add(self): """Add a notebook.""" id_ = self.api.add_notebook() @@ -390,7 +334,7 @@ def test_move(self): self.assertEqual(self.api.get_notebook(id_=second_id).parent_id, "") -class Ping(TestBase): +class Ping(ClientBase): def test_ping(self): """Ping should return the test string.""" ping = self.api.ping() @@ -405,7 +349,7 @@ def test_ping_wrong_method(self): self.assertEqual(context.exception.response.status_code, 405) -class Resource(TestBase): +class Resource(ClientBase): @with_resource def test_add(self, filename): """Add a resource.""" @@ -534,7 +478,7 @@ def test_check_property_title(self, filename): self.assertEqual(resource.title, title) -class Revision(TestBase): +class Revision(ClientBase): def test_add(self): """Add a revision.""" self.api.add_notebook() # notebook for the note @@ -591,7 +535,7 @@ def test_get_revisions_valid_properties(self): # TODO: Add more tests for the search parameter. -class Search(TestBase): +class Search(ClientBase): def test_empty(self): """Search should succeed, even if there is no result item.""" self.assertEqual(self.api.search(query="*").items, []) @@ -707,7 +651,7 @@ def test_search_all(self): self.assertEqual(len(self.api.search_all(**query)), count) -class Tag(TestBase): +class Tag(ClientBase): def test_add_no_note(self): """Tags can be added even without notes.""" id_ = self.api.add_tag() @@ -804,10 +748,10 @@ def test_get_tags_valid_properties(self): self.assertEqual(tag.assigned_fields(), set(properties)) -class Fuzz(TestBase): +class Fuzz(ClientBase): def test_random_path(self): """API should not crash, even with invalid paths.""" - for _ in range(1000 if SLOW_TESTS else 10): + for _ in range(1000 if common.SLOW_TESTS else 10): path = "/" + self.get_random_string(length=random.randint(0, 300)) method = random.choice(("delete", "get", "post", "put")) try: @@ -820,7 +764,7 @@ def test_random_path(self): self.api.ping() -class Helper(TestBase): +class Helper(ClientBase): """Tests for the helper functions.""" def test_random_id(self): @@ -839,7 +783,7 @@ def test_is_id_valid(self): self.assertFalse(dt.is_id_valid("h" + "0" * 31)) -class Miscellaneous(TestBase): +class Miscellaneous(ClientBase): @with_resource def test_same_id_different_type(self, filename): """Same IDs can be used if the types are different.""" @@ -851,9 +795,9 @@ def test_same_id_different_type(self, filename): self.api.add_tag(id_=id_) -class Regression(TestBase): +class Regression(ClientBase): @unittest.skip("Enable when the bug is fixed.") - @unittest.skipIf(not SLOW_TESTS, "Generating the long string is slow.") + @unittest.skipIf(not common.SLOW_TESTS, "Generating the long string is slow.") def test_long_body(self): """ https://github.com/laurent22/joplin/issues/5543 @@ -884,7 +828,7 @@ def test_add_todo(self): """https://github.com/laurent22/joplin/issues/1687""" -class ReadmeExamples(TestBase): +class ReadmeExamples(ClientBase): """Check the readme examples for functionality.""" readme_content: str = "" diff --git a/test/test_server_api.py b/test/test_server_api.py new file mode 100644 index 0000000..5cb0ee3 --- /dev/null +++ b/test/test_server_api.py @@ -0,0 +1,354 @@ +"""Tests for the Joplin server python API.""" + +from typing import cast + +from joppy.server_api import ServerApi +import joppy.data_types as dt +from . import common, setup_joplin + + +API = None +SERVER = None + + +def setUpModule(): # pylint: disable=invalid-name + global API, SERVER + SERVER = setup_joplin.JoplinServer() + # login only once to prevent + # "429 Client Error: Too Many Requests for url: http://localhost:22300/login" + API = ServerApi() + + +def tearDownModule(): # pylint: disable=invalid-name + if SERVER is not None: + SERVER.stop() + + +class ServerBase(common.Base): + def setUp(self): + self.api = cast(ServerApi, API) + super().setUp() + + +class Note(ServerBase): + def test_add(self): + """Add a note to an existing notebook.""" + parent_id = self.api.add_notebook() + id_ = self.api.add_note(parent_id=parent_id) + + notes = self.api.get_notes().items + self.assertEqual(len(notes), 1) + self.assertEqual(notes[0].id, id_) + self.assertEqual(notes[0].parent_id, parent_id) + + # TODO + # def test_add_attach_image(self): + # """Add a note with an image attached.""" + # self.api.add_notebook() + # image_data = tools.encode_base64("test/grant_authorization_button.png") + # note_id = self.api.add_note( + # image_data_url=f"data:image/png;base64,{image_data}" + # ) + + # # Check that there is a resource. + # resources = self.api.get_resources().items + # self.assertEqual(len(resources), 1) + # resource_id = resources[0].id + + # # Verify the resource is attached to the note. + # resources = self.api.get_resources(note_id=note_id).items + # self.assertEqual(len(resources), 1) + # self.assertEqual(resources[0].id, resource_id) + + def test_delete(self): + """Add and then delete a note.""" + self.api.add_notebook() + id_ = self.api.add_note() + notes = self.api.get_notes() + self.assertEqual(len(notes.items), 1) + + self.api.delete_note(id_=id_) + self.assertEqual(self.api.get_notes().items, []) + + def test_get_note(self): + """Get a specific note.""" + parent_id = self.api.add_notebook() + id_ = self.api.add_note(parent_id=parent_id) + note = self.api.get_note(id_=id_) + self.assertEqual(note.type_, dt.ItemType.NOTE) + + def test_get_notes(self): + """Get all notes.""" + parent_id = self.api.add_notebook() + self.api.add_note(parent_id=parent_id) + notes = self.api.get_notes() + self.assertEqual(len(notes.items), 1) + self.assertFalse(notes.has_more) + + def test_get_all_notes(self): + """Get all notes, unpaginated.""" + self.api.add_notebook() + count = 20 + for _ in range(count): + self.api.add_note() + self.assertEqual(len(self.api.get_all_notes()), count) + + def test_move(self): + """Move a note from one notebook to another.""" + original_title = "test title" + original_body = "test body" + + notebook_1_id = self.api.add_notebook() + notebook_2_id = self.api.add_notebook() + id_ = self.api.add_note( + parent_id=notebook_1_id, title=original_title, body=original_body + ) + + note = self.api.get_note(id_=id_) + self.assertEqual(note.parent_id, notebook_1_id) + + self.api.modify_note(id_=id_, parent_id=notebook_2_id) + note = self.api.get_note(id_=id_) + self.assertEqual(note.parent_id, notebook_2_id) + + # Ensure the original properties aren't modified. + self.assertEqual(note.body, original_body) + self.assertEqual(note.title, original_title) + + +class Notebook(ServerBase): + def test_add(self): + """Add a notebook.""" + id_ = self.api.add_notebook() + + notebooks = self.api.get_notebooks().items + self.assertEqual(len(notebooks), 1) + self.assertEqual(notebooks[0].id, id_) + + def test_get_notebook(self): + """Get a specific notebook.""" + id_ = self.api.add_notebook() + notebook = self.api.get_notebook(id_=id_) + self.assertEqual(notebook.type_, dt.ItemType.FOLDER) + + def test_get_notebooks(self): + """Get all notebooks.""" + self.api.add_notebook() + notebooks = self.api.get_notebooks() + self.assertEqual(len(notebooks.items), 1) + self.assertFalse(notebooks.has_more) + + def test_get_all_notebooks(self): + """Get all notebooks, unpaginated.""" + count = 20 + for _ in range(count): + self.api.add_notebook() + self.assertEqual(len(self.api.get_all_notebooks()), count) + + def test_move(self): + """Move a root notebok to another notebook. It's now a subnotebook.""" + first_id = self.api.add_notebook() + second_id = self.api.add_notebook() + self.assertEqual(self.api.get_notebook(id_=first_id).parent_id, None) + self.assertEqual(self.api.get_notebook(id_=second_id).parent_id, None) + + self.api.modify_notebook(id_=first_id, parent_id=second_id) + self.assertEqual(self.api.get_notebook(id_=first_id).parent_id, second_id) + self.assertEqual(self.api.get_notebook(id_=second_id).parent_id, None) + + +class Ping(ServerBase): + def test_ping(self): + """Ping should return the test string.""" + ping = self.api.ping() + self.assertEqual( + ping.json(), {"status": "ok", "message": "Joplin Server is running"} + ) + + +# TODO +# class Resource(ServerBase): +# @with_resource +# def test_add(self, filename): +# """Add a resource.""" +# id_ = self.api.add_resource(filename=filename) + +# resources = self.api.get_resources().items +# self.assertEqual(len(resources), 1) +# self.assertEqual(resources[0].id, id_) + +# @with_resource +# def test_add_to_note(self, filename): +# """Add a resource to an existing note.""" +# self.api.add_notebook() +# for file_ in ["test/grant_authorization_button.png", filename]: +# with self.subTest(file_=file_): +# note_id = self.api.add_note() +# resource_id = self.api.add_resource(filename=file_) +# self.api.add_resource_to_note(resource_id=resource_id, +# note_id=note_id) + +# # Verify the resource is attached to the note. +# resources = self.api.get_resources( +# note_id=note_id, fields="id,mime" +# ).items +# self.assertEqual(len(resources), 1) +# self.assertEqual(resources[0].id, resource_id) + +# # Verify the markdown is correct (prefix "!" for images). +# note = self.api.get_note(id_=note_id, fields="body") +# # TODO: Use "assertIsNotNone()" when +# # https://github.com/python/mypy/issues/5528 is resolved. +# assert resources[0].mime is not None +# image_prefix = "!" if resources[0].mime.startswith("image/") else "" +# self.assertEqual( +# f"\n{image_prefix}[{file_}](:/{resource_id})", note.body +# ) + +# @with_resource +# def test_delete(self, filename): +# """Add and then delete a resource.""" +# id_ = self.api.add_resource(filename=filename) +# resources = self.api.get_resources() +# self.assertEqual(len(resources.items), 1) + +# self.api.delete_resource(id_=id_) +# self.assertEqual(self.api.get_resources().items, []) +# self.assertEqual(os.listdir(f"{PROFILE}/resources"), []) + +# @with_resource +# def test_get_resource(self, filename): +# """Get metadata about a specific resource.""" +# id_ = self.api.add_resource(filename=filename) +# resource = self.api.get_resource(id_=id_) +# self.assertEqual(resource.assigned_fields(), resource.default_fields()) +# self.assertEqual(resource.type_, dt.ItemType.RESOURCE) + +# @with_resource +# def test_get_resource_file(self, filename): +# """Get a specific resource in binary format.""" +# for file_ in ("test/grant_authorization_button.png", filename): +# id_ = self.api.add_resource(filename=file_) +# resource = self.api.get_resource_file(id_=id_) +# with open(file_, "rb") as resource_file: +# self.assertEqual(resource, resource_file.read()) + +# @with_resource +# def test_get_resources(self, filename): +# """Get all resources.""" +# self.api.add_resource(filename=filename) +# resources = self.api.get_resources() +# self.assertEqual(len(resources.items), 1) +# self.assertFalse(resources.has_more) +# for resource in resources.items: +# self.assertEqual(resource.assigned_fields(), resource.default_fields()) + +# @with_resource +# def test_get_all_resources(self, filename): +# """Get all resources, unpaginated.""" +# # Small limit and count to create/remove as less as possible items. +# count, limit = random.randint(1, 10), random.randint(1, 10) +# for _ in range(count): +# self.api.add_resource(filename=filename) +# self.assertEqual( +# len(self.api.get_resources(limit=limit).items), min(limit, count) +# ) +# self.assertEqual(len(self.api.get_all_resources(limit=limit)), count) + +# @with_resource +# def test_get_resources_valid_properties(self, filename): +# """Try to get specific properties of a resource.""" +# self.api.add_resource(filename=filename) +# property_combinations = self.get_combinations(dt.ResourceData.fields()) +# for properties in property_combinations: +# resources = self.api.get_resources(fields=",".join(properties)) +# for resource in resources.items: +# self.assertEqual(resource.assigned_fields(), set(properties)) + +# @with_resource +# def test_modify_title(self, filename): +# """Modify a resource title.""" +# id_ = self.api.add_resource(filename=filename) + +# new_title = self.get_random_string() +# self.api.modify_resource(id_=id_, title=new_title) +# self.assertEqual(self.api.get_resource(id_=id_).title, new_title) + +# @with_resource +# def test_check_derived_properties(self, filename): +# """Check the derived properties. I. e. mime type, extension and size.""" +# for file_ in ["test/grant_authorization_button.png", filename]: +# id_ = self.api.add_resource(filename=file_) +# resource = self.api.get_resource(id_=id_, +# fields="mime,file_extension,size") +# mime_type, _ = mimetypes.guess_type(file_) +# self.assertEqual( +# resource.mime, +# mime_type if mime_type is not None else "application/octet-stream", +# ) +# self.assertEqual(resource.file_extension, os.path.splitext(file_)[1][1:]) +# self.assertEqual(resource.size, os.path.getsize(file_)) + +# @with_resource +# def test_check_property_title(self, filename): +# """Check the title of a resource.""" +# title = self.get_random_string() +# id_ = self.api.add_resource(filename=filename, title=title) +# resource = self.api.get_resource(id_=id_) +# self.assertEqual(resource.title, title) + + +class Tag(ServerBase): + def test_add_no_note(self): + """Tags can be added even without notes.""" + id_ = self.api.add_tag() + + tags = self.api.get_tags().items + self.assertEqual(len(tags), 1) + self.assertEqual(tags[0].id, id_) + + def test_add_to_note(self): + """Add a tag to an existing note.""" + self.api.add_notebook() + note_id = self.api.add_note() + tag_id = self.api.add_tag() + self.api.add_tag_to_note(tag_id=tag_id, note_id=note_id) + + # TODO: get_note_tag() + # notes = self.api.get_notes(tag_id=tag_id).items + # self.assertEqual(len(notes), 1) + # self.assertEqual(notes[0].id, note_id) + tags = self.api.get_all_tags() + self.assertEqual(len(tags), 1) + + def test_add_with_parent(self): + """Add a tag as child for an existing note.""" + self.api.add_notebook() + parent_id = self.api.add_note() + id_ = self.api.add_tag(parent_id=parent_id) + + tags = self.api.get_tags().items + self.assertEqual(len(tags), 1) + self.assertEqual(tags[0].id, id_) + self.assertEqual(tags[0].parent_id, parent_id) + + def test_get_tag(self): + """Get a specific tag.""" + id_ = self.api.add_tag() + tag = self.api.get_tag(id_=id_) + self.assertEqual(tag.type_, dt.ItemType.TAG) + + def test_get_tags(self): + """Get all tags.""" + self.api.add_tag() + tags = self.api.get_tags() + self.assertEqual(len(tags.items), 1) + self.assertFalse(tags.has_more) + + def test_get_all_tags(self): + """Get all tags, unpaginated.""" + # Small limit and count to create/remove as less as possible items. + count = 20 + for _ in range(count): + self.api.add_tag() + self.assertEqual(len(self.api.get_all_tags()), count)