mirror of
https://github.com/natelandau/obsidian-metadata.git
synced 2025-11-16 08:53:48 -05:00
refactor(application): refactor questions to separate class (#7)
* refactor(application): refactor questions to separate class * test(application): add tests for`Application` class
This commit is contained in:
Nathaniel Landau
@@ -5,14 +5,51 @@ from pathlib import Path
|
||||
from textwrap import dedent
|
||||
from typing import Any
|
||||
|
||||
import questionary
|
||||
import rich.repr
|
||||
import tomlkit
|
||||
import typer
|
||||
|
||||
from obsidian_metadata._utils import Questions, alerts
|
||||
from obsidian_metadata._utils import alerts
|
||||
from obsidian_metadata._utils.alerts import logger as log
|
||||
|
||||
|
||||
class ConfigQuestions:
|
||||
"""Questions to ask the user when creating a configuration file."""
|
||||
|
||||
@staticmethod
|
||||
def ask_for_vault_path() -> Path: # pragma: no cover
|
||||
"""Ask the user for the path to their vault.
|
||||
|
||||
Returns:
|
||||
Path: The path to the vault.
|
||||
"""
|
||||
vault_path = questionary.path(
|
||||
"Enter a path to Obsidian vault:",
|
||||
only_directories=True,
|
||||
validate=ConfigQuestions._validate_valid_dir,
|
||||
).ask()
|
||||
if vault_path is None:
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
return Path(vault_path).expanduser().resolve()
|
||||
|
||||
@staticmethod
|
||||
def _validate_valid_dir(path: str) -> bool | str:
|
||||
"""Validates a valid directory.
|
||||
|
||||
Returns:
|
||||
bool | str: True if the path is valid, otherwise a string with the error message.
|
||||
"""
|
||||
path_to_validate: Path = Path(path).expanduser().resolve()
|
||||
if not path_to_validate.exists():
|
||||
return f"Path does not exist: {path_to_validate}"
|
||||
if not path_to_validate.is_dir():
|
||||
return f"Path is not a directory: {path_to_validate}"
|
||||
|
||||
return True
|
||||
|
||||
|
||||
@rich.repr.auto
|
||||
class Config:
|
||||
"""Representation of a configuration file."""
|
||||
@@ -70,7 +107,7 @@ class Config:
|
||||
|
||||
def _write_default_config(self, path_to_config: Path) -> None:
|
||||
"""Write the default configuration file when no config file is found."""
|
||||
vault_path = Questions.ask_for_vault_path()
|
||||
vault_path = ConfigQuestions.ask_for_vault_path()
|
||||
|
||||
config_text = f"""\
|
||||
# Add another vault by replicating this section and changing the name
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
|
||||
from obsidian_metadata._utils import alerts
|
||||
from obsidian_metadata._utils.alerts import LoggerManager
|
||||
from obsidian_metadata._utils.questions import Questions
|
||||
from obsidian_metadata._utils.utilities import (
|
||||
clean_dictionary,
|
||||
clear_screen,
|
||||
@@ -21,7 +20,6 @@ __all__ = [
|
||||
"dict_contains",
|
||||
"docstring_parameter",
|
||||
"LoggerManager",
|
||||
"Questions",
|
||||
"remove_markdown_sections",
|
||||
"vault_validation",
|
||||
"version_callback",
|
||||
|
||||
@@ -1,33 +0,0 @@
|
||||
"""Functions for asking questions to the user and validating responses."""
|
||||
from pathlib import Path
|
||||
|
||||
import questionary
|
||||
import typer
|
||||
|
||||
|
||||
class Questions:
|
||||
"""Class for asking questions to the user and validating responses."""
|
||||
|
||||
@staticmethod
|
||||
def ask_for_vault_path() -> Path: # pragma: no cover
|
||||
"""Ask the user for the path to their vault."""
|
||||
vault_path = questionary.path(
|
||||
"Enter a path to Obsidian vault:",
|
||||
only_directories=True,
|
||||
validate=Questions._validate_vault,
|
||||
).ask()
|
||||
if vault_path is None:
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
return Path(vault_path).expanduser().resolve()
|
||||
|
||||
@staticmethod
|
||||
def _validate_vault(path: str) -> bool | str:
|
||||
"""Validates the vault path."""
|
||||
path_to_validate: Path = Path(path).expanduser().resolve()
|
||||
if not path_to_validate.exists():
|
||||
return f"Path does not exist: {path_to_validate}"
|
||||
if not path_to_validate.is_dir():
|
||||
return f"Path is not a directory: {path_to_validate}"
|
||||
|
||||
return True
|
||||
@@ -7,9 +7,10 @@ import questionary
|
||||
from rich import print
|
||||
|
||||
from obsidian_metadata._config import VaultConfig
|
||||
from obsidian_metadata._utils import alerts
|
||||
from obsidian_metadata._utils.alerts import logger as log
|
||||
from obsidian_metadata.models import Patterns, Vault
|
||||
from obsidian_metadata._utils import alerts
|
||||
from obsidian_metadata.models.questions import Questions
|
||||
|
||||
PATTERNS = Patterns()
|
||||
|
||||
@@ -25,14 +26,7 @@ class Application:
|
||||
def __init__(self, config: VaultConfig, dry_run: bool) -> None:
|
||||
self.config = config
|
||||
self.dry_run = dry_run
|
||||
self.custom_style = questionary.Style(
|
||||
[
|
||||
("separator", "bold fg:#6C6C6C"),
|
||||
("instruction", "fg:#6C6C6C"),
|
||||
("highlighted", "bold reverse"),
|
||||
("pointer", "bold"),
|
||||
]
|
||||
)
|
||||
self.questions = Questions()
|
||||
|
||||
def load_vault(self, path_filter: str = None) -> None:
|
||||
"""Load the vault.
|
||||
@@ -42,98 +36,52 @@ class Application:
|
||||
"""
|
||||
self.vault: Vault = Vault(config=self.config, dry_run=self.dry_run, path_filter=path_filter)
|
||||
log.info(f"Indexed {self.vault.num_notes()} notes from {self.vault.vault_path}")
|
||||
self.questions = Questions(vault=self.vault)
|
||||
|
||||
def main_app(self) -> None: # noqa: C901
|
||||
def main_app(self) -> None:
|
||||
"""Questions for the main application."""
|
||||
self.load_vault()
|
||||
|
||||
while True:
|
||||
print("\n")
|
||||
self.vault.info()
|
||||
operation = questionary.select(
|
||||
"What do you want to do?",
|
||||
choices=[
|
||||
questionary.Separator("\n-- VAULT ACTIONS -----------------"),
|
||||
{"name": "Backup vault", "value": "backup_vault"},
|
||||
{"name": "Delete vault backup", "value": "delete_backup"},
|
||||
{"name": "View all metadata", "value": "all_metadata"},
|
||||
{"name": "List notes in scope", "value": "list_notes"},
|
||||
{
|
||||
"name": "Filter the notes being processed by their path",
|
||||
"value": "filter_notes",
|
||||
},
|
||||
questionary.Separator("\n-- INLINE TAG ACTIONS ---------"),
|
||||
questionary.Separator("Tags in the note body"),
|
||||
{
|
||||
"name": "Rename an inline tag",
|
||||
"value": "rename_inline_tag",
|
||||
},
|
||||
{
|
||||
"name": "Delete an inline tag",
|
||||
"value": "delete_inline_tag",
|
||||
},
|
||||
questionary.Separator("\n-- METADATA ACTIONS -----------"),
|
||||
questionary.Separator("Frontmatter or inline metadata"),
|
||||
{"name": "Rename Key", "value": "rename_key"},
|
||||
{"name": "Delete Key", "value": "delete_key"},
|
||||
{"name": "Rename Value", "value": "rename_value"},
|
||||
{"name": "Delete Value", "value": "delete_value"},
|
||||
questionary.Separator("\n-- REVIEW/COMMIT CHANGES ------"),
|
||||
{"name": "Review changes", "value": "review_changes"},
|
||||
{"name": "Commit changes", "value": "commit_changes"},
|
||||
questionary.Separator("-------------------------------"),
|
||||
{"name": "Quit", "value": "abort"},
|
||||
],
|
||||
use_shortcuts=False,
|
||||
style=self.custom_style,
|
||||
).ask()
|
||||
|
||||
if operation == "filter_notes":
|
||||
path_filter = questionary.text(
|
||||
"Enter a regex to filter notes by path",
|
||||
validate=lambda text: len(text) > 0,
|
||||
).ask()
|
||||
if path_filter is None:
|
||||
continue
|
||||
self.load_vault(path_filter=path_filter)
|
||||
match self.questions.ask_main_application(): # noqa: E999
|
||||
case None:
|
||||
break
|
||||
case "filter_notes":
|
||||
self.load_vault(path_filter=self.questions.ask_for_filter_path())
|
||||
case "all_metadata":
|
||||
self.vault.metadata.print_metadata()
|
||||
case "backup_vault":
|
||||
self.vault.backup()
|
||||
case "delete_backup":
|
||||
self.vault.delete_backup()
|
||||
case "list_notes":
|
||||
self.vault.list_editable_notes()
|
||||
case "rename_inline_tag":
|
||||
self.rename_inline_tag()
|
||||
case "delete_inline_tag":
|
||||
self.delete_inline_tag()
|
||||
case "rename_key":
|
||||
self.rename_key()
|
||||
case "delete_key":
|
||||
self.delete_key()
|
||||
case "rename_value":
|
||||
self.rename_value()
|
||||
case "delete_value":
|
||||
self.delete_value()
|
||||
case "review_changes":
|
||||
self.review_changes()
|
||||
case "commit_changes":
|
||||
if self.commit_changes():
|
||||
break
|
||||
|
||||
if operation == "all_metadata":
|
||||
self.vault.metadata.print_metadata()
|
||||
log.error("Commit failed. Please run with -vvv for more info.")
|
||||
break
|
||||
|
||||
if operation == "backup_vault":
|
||||
self.vault.backup()
|
||||
|
||||
if operation == "delete_backup":
|
||||
self.vault.delete_backup()
|
||||
|
||||
if operation == "list_notes":
|
||||
self.vault.list_editable_notes()
|
||||
|
||||
if operation == "rename_inline_tag":
|
||||
self.rename_inline_tag()
|
||||
|
||||
if operation == "delete_inline_tag":
|
||||
self.delete_inline_tag()
|
||||
|
||||
if operation == "rename_key":
|
||||
self.rename_key()
|
||||
|
||||
if operation == "delete_key":
|
||||
self.delete_key()
|
||||
|
||||
if operation == "rename_value":
|
||||
self.rename_value()
|
||||
|
||||
if operation == "delete_value":
|
||||
self.delete_value()
|
||||
|
||||
if operation == "review_changes":
|
||||
self.review_changes()
|
||||
|
||||
if operation == "commit_changes" and self.commit_changes():
|
||||
break
|
||||
|
||||
if operation == "abort":
|
||||
break
|
||||
case "abort":
|
||||
break
|
||||
|
||||
print("Done!")
|
||||
return
|
||||
@@ -141,170 +89,126 @@ class Application:
|
||||
def rename_key(self) -> None:
|
||||
"""Renames a key in the vault."""
|
||||
|
||||
def validate_key(text: str) -> bool:
|
||||
"""Validate the key name."""
|
||||
if self.vault.metadata.contains(text):
|
||||
return True
|
||||
return False
|
||||
|
||||
def validate_new_key(text: str) -> bool:
|
||||
"""Validate the tag name."""
|
||||
if PATTERNS.validate_key_text.search(text) is not None:
|
||||
return False
|
||||
if len(text) == 0:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
original_key = questionary.text(
|
||||
"Which key would you like to rename?",
|
||||
validate=validate_key,
|
||||
).ask()
|
||||
original_key = self.questions.ask_for_existing_key(
|
||||
question="Which key would you like to rename?"
|
||||
)
|
||||
if original_key is None:
|
||||
return
|
||||
|
||||
new_key = questionary.text(
|
||||
"New key name",
|
||||
validate=validate_new_key,
|
||||
).ask()
|
||||
new_key = self.questions.ask_for_new_key()
|
||||
if new_key is None:
|
||||
return
|
||||
|
||||
self.vault.rename_metadata(original_key, new_key)
|
||||
num_changed = self.vault.rename_metadata(original_key, new_key)
|
||||
if num_changed == 0:
|
||||
alerts.warning(f"No notes were changed")
|
||||
return
|
||||
|
||||
alerts.success(
|
||||
f"Renamed [reverse]{original_key}[/] to [reverse]{new_key}[/] in {num_changed} notes"
|
||||
)
|
||||
|
||||
def rename_inline_tag(self) -> None:
|
||||
"""Rename an inline tag."""
|
||||
|
||||
def validate_new_tag(text: str) -> bool:
|
||||
"""Validate the tag name."""
|
||||
if PATTERNS.validate_tag_text.search(text) is not None:
|
||||
return False
|
||||
if len(text) == 0:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
original_tag = questionary.text(
|
||||
"Which tag would you like to rename?",
|
||||
validate=lambda text: True
|
||||
if self.vault.contains_inline_tag(text)
|
||||
else "Tag not found in vault",
|
||||
).ask()
|
||||
original_tag = self.questions.ask_for_existing_inline_tag(question="Which tag to rename?")
|
||||
if original_tag is None:
|
||||
return
|
||||
|
||||
new_tag = questionary.text(
|
||||
"New tag name",
|
||||
validate=validate_new_tag,
|
||||
).ask()
|
||||
new_tag = self.questions.ask_for_new_tag("New tag")
|
||||
if new_tag is None:
|
||||
return
|
||||
|
||||
self.vault.rename_inline_tag(original_tag, new_tag)
|
||||
alerts.success(f"Renamed [reverse]{original_tag}[/] to [reverse]{new_tag}[/]")
|
||||
num_changed = self.vault.rename_inline_tag(original_tag, new_tag)
|
||||
if num_changed == 0:
|
||||
alerts.warning(f"No notes were changed")
|
||||
return
|
||||
|
||||
alerts.success(
|
||||
f"Renamed [reverse]{original_tag}[/] to [reverse]{new_tag}[/] in {num_changed} notes"
|
||||
)
|
||||
return
|
||||
|
||||
def delete_inline_tag(self) -> None:
|
||||
"""Delete an inline tag."""
|
||||
tag = questionary.text(
|
||||
"Which tag would you like to delete?",
|
||||
validate=lambda text: True
|
||||
if self.vault.contains_inline_tag(text)
|
||||
else "Tag not found in vault",
|
||||
).ask()
|
||||
if tag is None:
|
||||
tag = self.questions.ask_for_existing_inline_tag(
|
||||
question="Which tag would you like to delete?"
|
||||
)
|
||||
|
||||
num_changed = self.vault.delete_inline_tag(tag)
|
||||
if num_changed == 0:
|
||||
alerts.warning(f"No notes were changed")
|
||||
return
|
||||
|
||||
self.vault.delete_inline_tag(tag)
|
||||
alerts.success(f"Deleted inline tag: {tag}")
|
||||
alerts.success(f"Deleted inline tag: {tag} in {num_changed} notes")
|
||||
return
|
||||
|
||||
def delete_key(self) -> None:
|
||||
"""Delete a key from the vault."""
|
||||
while True:
|
||||
key_to_delete = questionary.text("Regex for the key(s) you'd like to delete?").ask()
|
||||
if key_to_delete is None:
|
||||
return
|
||||
key_to_delete = self.questions.ask_for_existing_keys_regex(
|
||||
question="Regex for the key(s) you'd like to delete?"
|
||||
)
|
||||
if key_to_delete is None:
|
||||
return
|
||||
|
||||
if not self.vault.metadata.contains(key_to_delete, is_regex=True):
|
||||
alerts.warning(f"No matching keys in the vault: {key_to_delete}")
|
||||
continue
|
||||
num_changed = self.vault.delete_metadata(key_to_delete)
|
||||
if num_changed == 0:
|
||||
alerts.warning(f"No notes found with a key matching: [reverse]{key_to_delete}[/]")
|
||||
return
|
||||
|
||||
num_changed = self.vault.delete_metadata(key_to_delete)
|
||||
if num_changed == 0:
|
||||
alerts.warning(f"No notes found matching: [reverse]{key_to_delete}[/]")
|
||||
return
|
||||
|
||||
alerts.success(
|
||||
f"Deleted keys matching: [reverse]{key_to_delete}[/] from {num_changed} notes"
|
||||
)
|
||||
break
|
||||
alerts.success(
|
||||
f"Deleted keys matching: [reverse]{key_to_delete}[/] from {num_changed} notes"
|
||||
)
|
||||
|
||||
return
|
||||
|
||||
def rename_value(self) -> None:
|
||||
"""Rename a value in the vault."""
|
||||
key = questionary.text(
|
||||
"Which key contains the value to rename?",
|
||||
validate=lambda text: True
|
||||
if self.vault.metadata.contains(text)
|
||||
else "Key not found in vault",
|
||||
).ask()
|
||||
key = self.questions.ask_for_existing_key(
|
||||
question="Which key contains the value to rename?"
|
||||
)
|
||||
if key is None:
|
||||
return
|
||||
|
||||
value = questionary.text(
|
||||
"Which value would you like to rename?",
|
||||
validate=lambda text: True
|
||||
if self.vault.metadata.contains(key, text)
|
||||
else f"Value not found in {key}",
|
||||
).ask()
|
||||
question_key = Questions(vault=self.vault, key=key)
|
||||
value = question_key.ask_for_existing_value(
|
||||
question="Which value would you like to rename?"
|
||||
)
|
||||
if value is None:
|
||||
return
|
||||
|
||||
new_value = questionary.text(
|
||||
"New value?",
|
||||
validate=lambda text: True
|
||||
if not self.vault.metadata.contains(key, text)
|
||||
else f"Value already exists in {key}",
|
||||
).ask()
|
||||
new_value = question_key.ask_for_new_value()
|
||||
if new_value is None:
|
||||
return
|
||||
|
||||
if self.vault.rename_metadata(key, value, new_value):
|
||||
alerts.success(f"Renamed [reverse]{key}: {value}[/] to [reverse]{key}: {new_value}[/]")
|
||||
num_changes = self.vault.rename_metadata(key, value, new_value)
|
||||
if num_changes == 0:
|
||||
alerts.warning(f"No notes were changed")
|
||||
return
|
||||
|
||||
alerts.success(f"Renamed '{key}:{value}' to '{key}:{new_value}' in {num_changes} notes")
|
||||
|
||||
def delete_value(self) -> None:
|
||||
"""Delete a value from the vault."""
|
||||
while True:
|
||||
key = questionary.text(
|
||||
"Which key contains the value to delete?",
|
||||
).ask()
|
||||
if key is None:
|
||||
return
|
||||
if not self.vault.metadata.contains(key, is_regex=True):
|
||||
alerts.warning(f"No keys in value match: {key}")
|
||||
continue
|
||||
break
|
||||
key = self.questions.ask_for_existing_key(
|
||||
question="Which key contains the value to delete?"
|
||||
)
|
||||
if key is None:
|
||||
return
|
||||
|
||||
while True:
|
||||
value = questionary.text(
|
||||
"Regex for the value to delete",
|
||||
).ask()
|
||||
if value is None:
|
||||
return
|
||||
if not self.vault.metadata.contains(key, value, is_regex=True):
|
||||
alerts.warning(f"No matching key value pairs found in the vault: {key}: {value}")
|
||||
continue
|
||||
questions2 = Questions(vault=self.vault, key=key)
|
||||
value = questions2.ask_for_existing_value_regex(question="Regex for the value to delete")
|
||||
if value is None:
|
||||
return
|
||||
|
||||
num_changed = self.vault.delete_metadata(key, value)
|
||||
if num_changed == 0:
|
||||
alerts.warning(f"No notes found matching: [reverse]{key}: {value}[/]")
|
||||
return
|
||||
num_changed = self.vault.delete_metadata(key, value)
|
||||
if num_changed == 0:
|
||||
alerts.warning(f"No notes found matching: {key}: {value}")
|
||||
return
|
||||
|
||||
alerts.success(
|
||||
f"Deleted {num_changed} entries matching: [reverse]{key}[/]: [reverse]{value}[/]"
|
||||
)
|
||||
|
||||
break
|
||||
alerts.success(
|
||||
f"Deleted value [reverse]{value}[/] from key [reverse]{key}[/] in {num_changed} notes"
|
||||
)
|
||||
|
||||
return
|
||||
|
||||
@@ -317,7 +221,9 @@ class Application:
|
||||
return
|
||||
|
||||
print(f"\nFound {len(changed_notes)} changed notes in the vault.\n")
|
||||
answer = questionary.confirm("View diffs of individual files?", default=False).ask()
|
||||
answer = self.questions.ask_confirm(
|
||||
question="View diffs of individual files?", default=False
|
||||
)
|
||||
if not answer:
|
||||
return
|
||||
|
||||
@@ -330,16 +236,14 @@ class Application:
|
||||
choices.append(_selection)
|
||||
|
||||
choices.append(questionary.Separator())
|
||||
choices.append({"name": "Return", "value": "skip"})
|
||||
choices.append({"name": "Return", "value": "return"})
|
||||
|
||||
while True:
|
||||
note_to_review = questionary.select(
|
||||
"Select a new to view the diff.",
|
||||
note_to_review = self.questions.ask_for_selection(
|
||||
choices=choices,
|
||||
use_shortcuts=False,
|
||||
style=self.custom_style,
|
||||
).ask()
|
||||
if note_to_review is None or note_to_review == "skip":
|
||||
question="Select a new to view the diff",
|
||||
)
|
||||
if note_to_review is None or note_to_review == "return":
|
||||
break
|
||||
changed_notes[note_to_review].print_diff()
|
||||
|
||||
|
||||
425
src/obsidian_metadata/models/questions.py
Normal file
425
src/obsidian_metadata/models/questions.py
Normal file
@@ -0,0 +1,425 @@
|
||||
"""Functions for asking questions to the user and validating responses.
|
||||
|
||||
This module contains wrappers around questionary to ask questions to the user and validate responses. Mocking questionary has proven very difficult. This functionality is separated from the main application logic to make it easier to test.
|
||||
|
||||
Progress towards testing questionary can be found on this issue:
|
||||
https://github.com/tmbo/questionary/issues/35
|
||||
"""
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import questionary
|
||||
import typer
|
||||
|
||||
from obsidian_metadata.models.patterns import Patterns
|
||||
from obsidian_metadata.models.vault import Vault
|
||||
|
||||
PATTERNS = Patterns()
|
||||
|
||||
|
||||
class Questions:
|
||||
"""Class for asking questions to the user and validating responses with questionary."""
|
||||
|
||||
@staticmethod
|
||||
def ask_for_vault_path() -> Path: # pragma: no cover
|
||||
"""Ask the user for the path to their vault.
|
||||
|
||||
Returns:
|
||||
Path: The path to the vault.
|
||||
"""
|
||||
vault_path = questionary.path(
|
||||
"Enter a path to Obsidian vault:",
|
||||
only_directories=True,
|
||||
validate=Questions._validate_valid_dir,
|
||||
).ask()
|
||||
if vault_path is None:
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
return Path(vault_path).expanduser().resolve()
|
||||
|
||||
@staticmethod
|
||||
def _validate_valid_dir(path: str) -> bool | str:
|
||||
"""Validates a valid directory.
|
||||
|
||||
Returns:
|
||||
bool | str: True if the path is valid, otherwise a string with the error message.
|
||||
"""
|
||||
path_to_validate: Path = Path(path).expanduser().resolve()
|
||||
if not path_to_validate.exists():
|
||||
return f"Path does not exist: {path_to_validate}"
|
||||
if not path_to_validate.is_dir():
|
||||
return f"Path is not a directory: {path_to_validate}"
|
||||
|
||||
return True
|
||||
|
||||
def __init__(self, vault: Vault = None, key: str = None) -> None:
|
||||
"""Initialize the class.
|
||||
|
||||
Args:
|
||||
vault_path (Path, optional): The path to the vault. Defaults to None.
|
||||
vault (Vault, optional): The vault object. Defaults to None.
|
||||
key (str, optional): The key to use when validating a key, value pair. Defaults to None.
|
||||
"""
|
||||
self.style = questionary.Style(
|
||||
[
|
||||
("separator", "bold fg:#6C6C6C"),
|
||||
("instruction", "fg:#6C6C6C"),
|
||||
("highlighted", "bold reverse"),
|
||||
("pointer", "bold"),
|
||||
]
|
||||
)
|
||||
self.vault = vault
|
||||
self.key = key
|
||||
|
||||
def ask_confirm(self, question: str, default: bool = True) -> bool: # pragma: no cover
|
||||
"""Ask the user to confirm an action.
|
||||
|
||||
Args:
|
||||
question (str): The question to ask.
|
||||
default (bool, optional): The default value. Defaults to True.
|
||||
|
||||
Returns:
|
||||
bool: True if the user confirms, otherwise False.
|
||||
"""
|
||||
return questionary.confirm(question, default=default, style=self.style).ask()
|
||||
|
||||
def ask_main_application(self) -> str: # pragma: no cover
|
||||
"""Selectable list for the main application interface.
|
||||
|
||||
Args:
|
||||
style (questionary.Style): The style to use for the question.
|
||||
|
||||
Returns:
|
||||
str: The selected application.
|
||||
"""
|
||||
return questionary.select(
|
||||
"What do you want to do?",
|
||||
choices=[
|
||||
questionary.Separator("\n-- VAULT ACTIONS -----------------"),
|
||||
{"name": "Backup vault", "value": "backup_vault"},
|
||||
{"name": "Delete vault backup", "value": "delete_backup"},
|
||||
{"name": "View all metadata", "value": "all_metadata"},
|
||||
{"name": "List notes in scope", "value": "list_notes"},
|
||||
{
|
||||
"name": "Filter the notes being processed by their path",
|
||||
"value": "filter_notes",
|
||||
},
|
||||
questionary.Separator("\n-- INLINE TAG ACTIONS ---------"),
|
||||
questionary.Separator("Tags in the note body"),
|
||||
{
|
||||
"name": "Rename an inline tag",
|
||||
"value": "rename_inline_tag",
|
||||
},
|
||||
{
|
||||
"name": "Delete an inline tag",
|
||||
"value": "delete_inline_tag",
|
||||
},
|
||||
questionary.Separator("\n-- METADATA ACTIONS -----------"),
|
||||
questionary.Separator("Frontmatter or inline metadata"),
|
||||
{"name": "Rename Key", "value": "rename_key"},
|
||||
{"name": "Delete Key", "value": "delete_key"},
|
||||
{"name": "Rename Value", "value": "rename_value"},
|
||||
{"name": "Delete Value", "value": "delete_value"},
|
||||
questionary.Separator("\n-- REVIEW/COMMIT CHANGES ------"),
|
||||
{"name": "Review changes", "value": "review_changes"},
|
||||
{"name": "Commit changes", "value": "commit_changes"},
|
||||
questionary.Separator("-------------------------------"),
|
||||
{"name": "Quit", "value": "abort"},
|
||||
],
|
||||
use_shortcuts=False,
|
||||
style=self.style,
|
||||
).ask()
|
||||
|
||||
def ask_for_filter_path(self) -> str: # pragma: no cover
|
||||
"""Ask the user for the path to the filter file.
|
||||
|
||||
Returns:
|
||||
str: The regex to use for filtering.
|
||||
"""
|
||||
filter_path_regex = questionary.path(
|
||||
"Regex to filter the notes being processed by their path:",
|
||||
only_directories=False,
|
||||
validate=self._validate_valid_vault_regex,
|
||||
).ask()
|
||||
if filter_path_regex is None:
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
return filter_path_regex
|
||||
|
||||
def ask_for_selection(
|
||||
self, choices: list[Any], question: str = "Select an option"
|
||||
) -> Any: # pragma: no cover
|
||||
"""Ask the user to select an item from a list.
|
||||
|
||||
Args:
|
||||
question (str, optional): The question to ask. Defaults to "Select an option".
|
||||
choices (list[Any]): The list of choices.
|
||||
|
||||
Returns:
|
||||
any: The selected item value.
|
||||
"""
|
||||
return questionary.select(
|
||||
"Select an item:",
|
||||
choices=choices,
|
||||
use_shortcuts=False,
|
||||
style=self.style,
|
||||
).ask()
|
||||
|
||||
def ask_for_existing_inline_tag(self, question: str = "Enter a tag") -> str: # pragma: no cover
|
||||
"""Ask the user for an existing inline tag."""
|
||||
return questionary.text(
|
||||
question,
|
||||
validate=self._validate_existing_inline_tag,
|
||||
).ask()
|
||||
|
||||
def ask_for_new_tag(self, question: str = "New tag name") -> str: # pragma: no cover
|
||||
"""Ask the user for a new inline tag."""
|
||||
return questionary.text(
|
||||
question,
|
||||
validate=self._validate_new_tag,
|
||||
).ask()
|
||||
|
||||
def ask_for_existing_key(self, question: str = "Enter a key") -> str: # pragma: no cover
|
||||
"""Ask the user for a metadata key.
|
||||
|
||||
Args:
|
||||
question (str, optional): The question to ask. Defaults to "Enter a key".
|
||||
|
||||
Returns:
|
||||
str: A metadata key that exists in the vault.
|
||||
"""
|
||||
return questionary.text(
|
||||
question,
|
||||
validate=self._validate_key_exists,
|
||||
).ask()
|
||||
|
||||
def ask_for_existing_keys_regex(
|
||||
self, question: str = "Regex for keys"
|
||||
) -> str: # pragma: no cover
|
||||
"""Ask the user for a regex for metadata keys.
|
||||
|
||||
Args:
|
||||
question (str, optional): The question to ask. Defaults to "Regex for keys".
|
||||
|
||||
Returns:
|
||||
str: A regex for metadata keys that exist in the vault.
|
||||
"""
|
||||
return questionary.text(
|
||||
question,
|
||||
validate=self._validate_key_exists_regex,
|
||||
).ask()
|
||||
|
||||
def ask_for_existing_value_regex(
|
||||
self, question: str = "Regex for values"
|
||||
) -> str: # pragma: no cover
|
||||
"""Ask the user for a regex for metadata values.
|
||||
|
||||
Args:
|
||||
question (str, optional): The question to ask. Defaults to "Regex for values".
|
||||
|
||||
Returns:
|
||||
str: A regex for metadata values that exist in the vault.
|
||||
"""
|
||||
return questionary.text(
|
||||
question,
|
||||
validate=self._validate_value_exists_regex,
|
||||
).ask()
|
||||
|
||||
def ask_for_existing_value(self, question: str = "Enter a value") -> str: # pragma: no cover
|
||||
"""Ask the user for a metadata value.
|
||||
|
||||
Args:
|
||||
question (str, optional): The question to ask. Defaults to "Enter a value".
|
||||
|
||||
Returns:
|
||||
str: A metadata value.
|
||||
"""
|
||||
return questionary.text(question, validate=self._validate_value).ask()
|
||||
|
||||
def ask_for_new_key(self, question: str = "New key name") -> str: # pragma: no cover
|
||||
"""Ask the user for a new metadata key.
|
||||
|
||||
Args:
|
||||
question (str, optional): The question to ask. Defaults to "New key name".
|
||||
|
||||
Returns:
|
||||
str: A new metadata key.
|
||||
"""
|
||||
return questionary.text(
|
||||
question,
|
||||
validate=self._validate_new_key,
|
||||
).ask()
|
||||
|
||||
def ask_for_new_value(self, question: str = "New value") -> str: # pragma: no cover
|
||||
"""Ask the user for a new metadata value.
|
||||
|
||||
Args:
|
||||
question (str, optional): The question to ask. Defaults to "New value".
|
||||
|
||||
Returns:
|
||||
str: A new metadata value.
|
||||
"""
|
||||
return questionary.text(
|
||||
question,
|
||||
validate=self._validate_new_value,
|
||||
).ask()
|
||||
|
||||
def _validate_key_exists(self, text: str) -> bool | str:
|
||||
"""Validates a valid key.
|
||||
|
||||
Returns:
|
||||
bool | str: True if the key is valid, otherwise a string with the error message.
|
||||
"""
|
||||
if len(text) < 1:
|
||||
return "Key cannot be empty"
|
||||
|
||||
if not self.vault.metadata.contains(text):
|
||||
return f"'{text}' does not exist as a key in the vault"
|
||||
|
||||
return True
|
||||
|
||||
def _validate_key_exists_regex(self, text: str) -> bool | str:
|
||||
"""Validates a valid key.
|
||||
|
||||
Returns:
|
||||
bool | str: True if the key is valid, otherwise a string with the error message.
|
||||
"""
|
||||
if len(text) < 1:
|
||||
return "Key cannot be empty"
|
||||
|
||||
try:
|
||||
re.compile(text)
|
||||
except re.error as error:
|
||||
return f"Invalid regex: {error}"
|
||||
|
||||
if not self.vault.metadata.contains(text, is_regex=True):
|
||||
return f"'{text}' does not exist as a key in the vault"
|
||||
|
||||
return True
|
||||
|
||||
def _validate_existing_inline_tag(self, text: str) -> bool | str:
|
||||
"""Validates an existing inline tag.
|
||||
|
||||
Returns:
|
||||
bool | str: True if the tag is valid, otherwise a string with the error message.
|
||||
"""
|
||||
if len(text) < 1:
|
||||
return "Tag cannot be empty"
|
||||
|
||||
if not self.vault.contains_inline_tag(text):
|
||||
return f"'{text}' does not exist as a tag in the vault"
|
||||
|
||||
return True
|
||||
|
||||
def _validate_valid_vault_regex(self, text: str) -> bool | str:
|
||||
"""Validates a valid regex.
|
||||
|
||||
Returns:
|
||||
bool | str: True if the regex is valid, otherwise a string with the error message.
|
||||
"""
|
||||
if len(text) < 1:
|
||||
return "Regex cannot be empty"
|
||||
|
||||
try:
|
||||
re.compile(text)
|
||||
except re.error as error:
|
||||
return f"Invalid regex: {error}"
|
||||
|
||||
if self.vault is not None:
|
||||
for subdir in list(self.vault.vault_path.glob("**/*")):
|
||||
if re.search(text, str(subdir)):
|
||||
return True
|
||||
return "Regex does not match paths in the vault"
|
||||
|
||||
return True
|
||||
|
||||
def _validate_new_key(self, text: str) -> bool | str:
|
||||
"""Validate the tag name.
|
||||
|
||||
Args:
|
||||
text (str): The key name to validate.
|
||||
|
||||
Returns:
|
||||
bool | str: True if the key is valid, otherwise a string with the error message.
|
||||
"""
|
||||
if PATTERNS.validate_key_text.search(text) is not None:
|
||||
return "Key cannot contain spaces or special characters"
|
||||
|
||||
if len(text) == 0:
|
||||
return "New key cannot be empty"
|
||||
|
||||
return True
|
||||
|
||||
def _validate_new_tag(self, text: str) -> bool | str:
|
||||
"""Validate the tag name.
|
||||
|
||||
Args:
|
||||
text (str): The tag name to validate.
|
||||
|
||||
Returns:
|
||||
bool | str: True if the tag is valid, otherwise a string with the error message.
|
||||
"""
|
||||
if PATTERNS.validate_tag_text.search(text) is not None:
|
||||
return "Tag cannot contain spaces or special characters"
|
||||
|
||||
if len(text) == 0:
|
||||
return "New tag cannot be empty"
|
||||
|
||||
return True
|
||||
|
||||
def _validate_value(self, text: str) -> bool | str:
|
||||
"""Validate the value.
|
||||
|
||||
Args:
|
||||
text (str): The value to validate.
|
||||
|
||||
Returns:
|
||||
bool | str: True if the value is valid, otherwise a string with the error message.
|
||||
"""
|
||||
if len(text) < 1:
|
||||
return "Value cannot be empty"
|
||||
|
||||
if self.key is not None and not self.vault.metadata.contains(self.key, text):
|
||||
return f"{self.key}:{text} does not exist"
|
||||
|
||||
return True
|
||||
|
||||
def _validate_value_exists_regex(self, text: str) -> bool | str:
|
||||
"""Validate the value.
|
||||
|
||||
Args:
|
||||
text (str): The value to validate.
|
||||
|
||||
Returns:
|
||||
bool | str: True if the value is valid, otherwise a string with the error message.
|
||||
"""
|
||||
if len(text) < 1:
|
||||
return "Regex cannot be empty"
|
||||
|
||||
try:
|
||||
re.compile(text)
|
||||
except re.error as error:
|
||||
return f"Invalid regex: {error}"
|
||||
|
||||
if self.key is not None and not self.vault.metadata.contains(self.key, text, is_regex=True):
|
||||
return f"No values in {self.key} match regex: {text}"
|
||||
|
||||
return True
|
||||
|
||||
def _validate_new_value(self, text: str) -> bool | str:
|
||||
"""Validate a new value.
|
||||
|
||||
Args:
|
||||
text (str): The value to validate.
|
||||
|
||||
Returns:
|
||||
bool | str: True if the value is valid, otherwise a string with the error message.
|
||||
"""
|
||||
if len(text) < 1:
|
||||
return "Value cannot be empty"
|
||||
|
||||
if self.key is not None and self.vault.metadata.contains(self.key, text):
|
||||
return f"{self.key}:{text} already exists"
|
||||
|
||||
return True
|
||||
@@ -146,25 +146,25 @@ class Vault:
|
||||
else:
|
||||
alerts.info("No backup found")
|
||||
|
||||
def delete_inline_tag(self, tag: str) -> bool:
|
||||
def delete_inline_tag(self, tag: str) -> int:
|
||||
"""Delete an inline tag in the vault.
|
||||
|
||||
Args:
|
||||
tag (str): Tag to delete.
|
||||
|
||||
Returns:
|
||||
bool: True if tag was deleted.
|
||||
int: Number of notes that had tag deleted.
|
||||
"""
|
||||
changes = False
|
||||
num_changed = 0
|
||||
|
||||
for _note in self.notes:
|
||||
if _note.delete_inline_tag(tag):
|
||||
changes = True
|
||||
num_changed += 1
|
||||
|
||||
if changes:
|
||||
if num_changed > 0:
|
||||
self.metadata.delete(self.notes[0].inline_tags.metadata_key, tag)
|
||||
return True
|
||||
return False
|
||||
|
||||
return num_changed
|
||||
|
||||
def delete_metadata(self, key: str, value: str = None) -> int:
|
||||
"""Delete metadata in the vault.
|
||||
@@ -184,7 +184,7 @@ class Vault:
|
||||
|
||||
if num_changed > 0:
|
||||
self.metadata.delete(key, value)
|
||||
return num_changed
|
||||
|
||||
return num_changed
|
||||
|
||||
def get_changed_notes(self) -> list[Note]:
|
||||
@@ -239,7 +239,7 @@ class Vault:
|
||||
"""
|
||||
return len(self.notes)
|
||||
|
||||
def rename_metadata(self, key: str, value_1: str, value_2: str = None) -> bool:
|
||||
def rename_metadata(self, key: str, value_1: str, value_2: str = None) -> int:
|
||||
"""Renames a key or key-value pair in the note's metadata.
|
||||
|
||||
If no value is provided, will rename an entire key.
|
||||
@@ -250,19 +250,20 @@ class Vault:
|
||||
value_2 (str, optional): New value.
|
||||
|
||||
Returns:
|
||||
bool: True if metadata was renamed.
|
||||
int: Number of notes that had metadata renamed.
|
||||
"""
|
||||
changes = False
|
||||
num_changed = 0
|
||||
|
||||
for _note in self.notes:
|
||||
if _note.rename_metadata(key, value_1, value_2):
|
||||
changes = True
|
||||
num_changed += 1
|
||||
|
||||
if changes:
|
||||
if num_changed > 0:
|
||||
self.metadata.rename(key, value_1, value_2)
|
||||
return True
|
||||
return False
|
||||
|
||||
def rename_inline_tag(self, old_tag: str, new_tag: str) -> bool:
|
||||
return num_changed
|
||||
|
||||
def rename_inline_tag(self, old_tag: str, new_tag: str) -> int:
|
||||
"""Rename an inline tag in the vault.
|
||||
|
||||
Args:
|
||||
@@ -270,17 +271,18 @@ class Vault:
|
||||
new_tag (str): New tag name.
|
||||
|
||||
Returns:
|
||||
bool: True if tag was renamed.
|
||||
int: Number of notes that had inline tags renamed.
|
||||
"""
|
||||
changes = False
|
||||
num_changed = 0
|
||||
|
||||
for _note in self.notes:
|
||||
if _note.rename_inline_tag(old_tag, new_tag):
|
||||
changes = True
|
||||
num_changed += 1
|
||||
|
||||
if changes:
|
||||
if num_changed > 0:
|
||||
self.metadata.rename(self.notes[0].inline_tags.metadata_key, old_tag, new_tag)
|
||||
return True
|
||||
return False
|
||||
|
||||
return num_changed
|
||||
|
||||
def write(self) -> None:
|
||||
"""Write changes to the vault."""
|
||||
|
||||
@@ -1,19 +1,414 @@
|
||||
# type: ignore
|
||||
"""Tests for the application module."""
|
||||
"""Tests for the application module.
|
||||
|
||||
How mocking works in this test suite:
|
||||
|
||||
1. The main_app() method is mocked using a side effect iterable. This allows us to pass a value in the first run, and then a KeyError in the second run to exit the loop.
|
||||
2. All questions are mocked using return_value. This allows us to pass in a value to the question and then the method will return that value. This is useful for testing questionary prompts without user input.
|
||||
"""
|
||||
|
||||
import re
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.helpers import Regex
|
||||
|
||||
|
||||
from obsidian_metadata._config import Config
|
||||
from obsidian_metadata.models.application import Application
|
||||
|
||||
|
||||
def test_load_vault(test_vault) -> None:
|
||||
def test_instantiate_application(test_application) -> None:
|
||||
"""Test application."""
|
||||
vault_path = test_vault
|
||||
config = Config(config_path="tests/fixtures/test_vault_config.toml", vault_path=vault_path)
|
||||
vault_config = config.vaults[0]
|
||||
app = Application(config=vault_config, dry_run=False)
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
|
||||
assert app.dry_run is False
|
||||
assert app.config == vault_config
|
||||
assert app.vault.num_notes() == 3
|
||||
assert app.config.name == "command_line_vault"
|
||||
assert app.config.exclude_paths == [".git", ".obsidian"]
|
||||
assert app.dry_run is False
|
||||
assert app.vault.num_notes() == 13
|
||||
|
||||
|
||||
def test_abort(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming a key."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
return_value="abort",
|
||||
)
|
||||
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert "Vault Info" in captured.out
|
||||
assert "Done!" in captured.out
|
||||
|
||||
|
||||
def test_list_notes(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming a key."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["list_notes", KeyError],
|
||||
)
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert "04 no metadata/no_metadata_1.md" in captured.out
|
||||
assert "02 inline/inline 2.md" in captured.out
|
||||
assert "+inbox/Untitled.md" in captured.out
|
||||
assert "00 meta/templates/data sample.md" in captured.out
|
||||
|
||||
|
||||
def test_all_metadata(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming a key."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["all_metadata", KeyError],
|
||||
)
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
expected = re.escape("┃ Keys ┃ Values")
|
||||
assert captured.out == Regex(expected)
|
||||
expected = re.escape("Inline Tags │ breakfast")
|
||||
assert captured.out == Regex(expected)
|
||||
|
||||
|
||||
def test_filter_notes(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming a key."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["filter_notes", "list_notes", KeyError],
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_filter_path",
|
||||
return_value="inline",
|
||||
)
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert "04 no metadata/no_metadata_1.md" not in captured.out
|
||||
assert "02 inline/inline 1.md" in captured.out
|
||||
assert "02 inline/inline 2.md" in captured.out
|
||||
assert "+inbox/Untitled.md" not in captured.out
|
||||
assert "00 meta/templates/data sample.md" not in captured.out
|
||||
|
||||
|
||||
def test_rename_key_success(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming a key."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["rename_key", KeyError],
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_key",
|
||||
return_value="tags",
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_new_key",
|
||||
return_value="new_tags",
|
||||
)
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == Regex(r"Renamed.*tags.*to.*new_tags.*in.*\d+.*notes", re.DOTALL)
|
||||
|
||||
|
||||
def test_rename_key_fail(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming a key."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["rename_key", KeyError],
|
||||
)
|
||||
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_key",
|
||||
return_value="tag",
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_new_key",
|
||||
return_value="new_tags",
|
||||
)
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert "WARNING | No notes were changed" in captured.out
|
||||
|
||||
|
||||
def test_rename_inline_tag_success(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming an inline tag."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["rename_inline_tag", KeyError],
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_inline_tag",
|
||||
return_value="breakfast",
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_new_tag",
|
||||
return_value="new_tag",
|
||||
)
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == Regex(r"Renamed.*breakfast.*to.*new_tag.*in.*\d+.*notes", re.DOTALL)
|
||||
|
||||
|
||||
def test_rename_inline_tag_fail(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming an inline tag."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["rename_inline_tag", KeyError],
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_inline_tag",
|
||||
return_value="not_a_tag",
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_new_tag",
|
||||
return_value="new_tag",
|
||||
)
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == Regex(r"WARNING +\| No notes were changed", re.DOTALL)
|
||||
|
||||
|
||||
def test_delete_inline_tag_success(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming an inline tag."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["delete_inline_tag", KeyError],
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_inline_tag",
|
||||
return_value="breakfast",
|
||||
)
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == Regex(r"SUCCESS +\| Deleted.*\d+.*notes", re.DOTALL)
|
||||
|
||||
|
||||
def test_delete_inline_tag_fail(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming an inline tag."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["delete_inline_tag", KeyError],
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_inline_tag",
|
||||
return_value="not_a_tag_in_vault",
|
||||
)
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == Regex(r"WARNING +\| No notes were changed", re.DOTALL)
|
||||
|
||||
|
||||
def test_delete_key_success(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming an inline tag."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["delete_key", KeyError],
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_keys_regex",
|
||||
return_value=r"d\w+",
|
||||
)
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == Regex(
|
||||
r"SUCCESS +\|.*Deleted.*keys.*matching:.*d\\w\+.*from.*10", re.DOTALL
|
||||
)
|
||||
|
||||
|
||||
def test_delete_key_fail(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming an inline tag."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["delete_key", KeyError],
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_keys_regex",
|
||||
return_value=r"\d{7}",
|
||||
)
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == Regex(r"WARNING +\| No notes found with a.*key.*matching", re.DOTALL)
|
||||
|
||||
|
||||
def test_rename_value_success(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming an inline tag."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["rename_value", KeyError],
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_key",
|
||||
return_value="area",
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_value",
|
||||
return_value="frontmatter",
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_new_value",
|
||||
return_value="new_key",
|
||||
)
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == Regex(
|
||||
r"SUCCESS | Renamed 'area:frontmatter' to 'area:new_key'", re.DOTALL
|
||||
)
|
||||
assert captured.out == Regex(r".*in.*\d+.*notes.*", re.DOTALL)
|
||||
|
||||
|
||||
def test_rename_value_fail(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming an inline tag."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["rename_value", KeyError],
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_key",
|
||||
return_value="area",
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_value",
|
||||
return_value="not_exists",
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_new_value",
|
||||
return_value="new_key",
|
||||
)
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == Regex(r"WARNING +\| No notes were changed", re.DOTALL)
|
||||
|
||||
|
||||
def test_delete_value_success(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming an inline tag."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["delete_value", KeyError],
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_key",
|
||||
return_value="area",
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_value_regex",
|
||||
return_value=r"^front\w+$",
|
||||
)
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == Regex(
|
||||
r"SUCCESS +\| Deleted value.*\^front\\w\+\$.*from.*key.*area.*in.*\d+.*notes", re.DOTALL
|
||||
)
|
||||
|
||||
|
||||
def test_delete_value_fail(test_application, mocker, capsys) -> None:
|
||||
"""Test renaming an inline tag."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["delete_value", KeyError],
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_key",
|
||||
return_value="area",
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_value_regex",
|
||||
return_value=r"\d{7}",
|
||||
)
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == Regex(r"WARNING +\| No notes found matching:", re.DOTALL)
|
||||
|
||||
|
||||
def test_review_no_changes(test_application, mocker, capsys) -> None:
|
||||
"""Review changes when no changes to vault."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["review_changes", KeyError],
|
||||
)
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == Regex(r"INFO +\| No changes to review", re.DOTALL)
|
||||
|
||||
|
||||
def test_review_changes(test_application, mocker, capsys) -> None:
|
||||
"""Review changes when no changes to vault."""
|
||||
app = test_application
|
||||
app.load_vault()
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_main_application",
|
||||
side_effect=["delete_key", "review_changes", KeyError],
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_existing_keys_regex",
|
||||
return_value=r"d\w+",
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_confirm",
|
||||
return_value=True,
|
||||
)
|
||||
mocker.patch(
|
||||
"obsidian_metadata.models.application.Questions.ask_for_selection",
|
||||
side_effect=[1, "return"],
|
||||
)
|
||||
with pytest.raises(KeyError):
|
||||
app.main_app()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == Regex(r".*Found.*\d+.*changed notes in the vault.*", re.DOTALL)
|
||||
assert "- date_created: 2022-12-22" in captured.out
|
||||
assert "+ - breakfast" in captured.out
|
||||
|
||||
@@ -7,7 +7,14 @@ from textwrap import dedent
|
||||
import pytest
|
||||
import typer
|
||||
|
||||
from obsidian_metadata._config import Config
|
||||
from obsidian_metadata._config.config import Config, ConfigQuestions
|
||||
|
||||
|
||||
def test_validate_valid_dir() -> None:
|
||||
"""Test vault validation."""
|
||||
assert ConfigQuestions._validate_valid_dir("tests/") is True
|
||||
assert "Path is not a directory" in ConfigQuestions._validate_valid_dir("pyproject.toml")
|
||||
assert "Path does not exist" in ConfigQuestions._validate_valid_dir("tests/vault2")
|
||||
|
||||
|
||||
def test_broken_config_file(capsys) -> None:
|
||||
@@ -79,8 +86,10 @@ def test_no_config_no_vault(tmp_path, mocker) -> None:
|
||||
"""Test creating a config on first run."""
|
||||
fake_vault = Path(tmp_path / "vault")
|
||||
fake_vault.mkdir()
|
||||
|
||||
mocker.patch(
|
||||
"obsidian_metadata._config.config.Questions.ask_for_vault_path", return_value=fake_vault
|
||||
"obsidian_metadata._config.config.ConfigQuestions.ask_for_vault_path",
|
||||
return_value=fake_vault,
|
||||
)
|
||||
|
||||
config_file = Path(tmp_path / "config.toml")
|
||||
|
||||
@@ -6,6 +6,9 @@ from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from obsidian_metadata._config import Config
|
||||
from obsidian_metadata.models.application import Application
|
||||
|
||||
|
||||
def remove_all(root: Path):
|
||||
"""Remove all files and directories in a directory."""
|
||||
@@ -72,3 +75,27 @@ def test_vault(tmp_path) -> Path:
|
||||
|
||||
if backup_dir.exists():
|
||||
shutil.rmtree(backup_dir)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def test_application(tmp_path) -> Application:
|
||||
"""Fixture which creates a sample vault."""
|
||||
source_dir = Path(__file__).parent / "fixtures" / "sample_vault"
|
||||
dest_dir = Path(tmp_path / "application")
|
||||
backup_dir = Path(f"{dest_dir}.bak")
|
||||
|
||||
if not source_dir.exists():
|
||||
raise FileNotFoundError(f"Sample vault not found: {source_dir}")
|
||||
|
||||
shutil.copytree(source_dir, dest_dir)
|
||||
config = Config(config_path="tests/fixtures/test_vault_config.toml", vault_path=dest_dir)
|
||||
vault_config = config.vaults[0]
|
||||
app = Application(config=vault_config, dry_run=False)
|
||||
|
||||
yield app
|
||||
|
||||
# after test - remove fixtures
|
||||
shutil.rmtree(dest_dir)
|
||||
|
||||
if backup_dir.exists():
|
||||
shutil.rmtree(backup_dir)
|
||||
|
||||
@@ -1,12 +1,113 @@
|
||||
# type: ignore
|
||||
"""Test the questions class."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from obsidian_metadata._utils import Questions
|
||||
from obsidian_metadata._config import Config
|
||||
from obsidian_metadata.models.questions import Questions
|
||||
from obsidian_metadata.models.vault import Vault
|
||||
|
||||
VAULT_PATH = Path("tests/fixtures/test_vault")
|
||||
CONFIG = Config(config_path="tests/fixtures/test_vault_config.toml", vault_path=VAULT_PATH)
|
||||
VAULT_CONFIG = CONFIG.vaults[0]
|
||||
VAULT = Vault(config=VAULT_CONFIG)
|
||||
|
||||
|
||||
def test_vault_validation():
|
||||
def test_validate_valid_dir() -> None:
|
||||
"""Test vault validation."""
|
||||
assert Questions._validate_vault("tests/") is True
|
||||
assert "Path is not a directory" in Questions._validate_vault("pyproject.toml")
|
||||
assert "Path does not exist" in Questions._validate_vault("tests/vault2")
|
||||
questions = Questions(vault=VAULT)
|
||||
assert questions._validate_valid_dir("tests/") is True
|
||||
assert "Path is not a directory" in questions._validate_valid_dir("pyproject.toml")
|
||||
assert "Path does not exist" in questions._validate_valid_dir("tests/vault2")
|
||||
|
||||
|
||||
def test_validate_valid_regex() -> None:
|
||||
"""Test regex validation."""
|
||||
questions = Questions(vault=VAULT)
|
||||
assert questions._validate_valid_vault_regex(r".*\.md") is True
|
||||
assert "Invalid regex" in questions._validate_valid_vault_regex("[")
|
||||
assert "Regex cannot be empty" in questions._validate_valid_vault_regex("")
|
||||
assert "Regex does not match paths" in questions._validate_valid_vault_regex(r"\d\d\d\w\d")
|
||||
|
||||
|
||||
def test_validate_key_exists() -> None:
|
||||
"""Test key validation."""
|
||||
questions = Questions(vault=VAULT)
|
||||
assert "'test' does not exist" in questions._validate_key_exists("test")
|
||||
assert "Key cannot be empty" in questions._validate_key_exists("")
|
||||
assert questions._validate_key_exists("frontmatter_Key1") is True
|
||||
|
||||
|
||||
def test_validate_new_key() -> None:
|
||||
"""Test new key validation."""
|
||||
questions = Questions(vault=VAULT)
|
||||
assert "Key cannot contain spaces or special characters" in questions._validate_new_key(
|
||||
"new key"
|
||||
)
|
||||
assert "Key cannot contain spaces or special characters" in questions._validate_new_key(
|
||||
"new_key!"
|
||||
)
|
||||
assert "New key cannot be empty" in questions._validate_new_key("")
|
||||
assert questions._validate_new_key("new_key") is True
|
||||
|
||||
|
||||
def test_validate_new_tag() -> None:
|
||||
"""Test new tag validation."""
|
||||
questions = Questions(vault=VAULT)
|
||||
assert "New tag cannot be empty" in questions._validate_new_tag("")
|
||||
assert "Tag cannot contain spaces or special characters" in questions._validate_new_tag(
|
||||
"new tag"
|
||||
)
|
||||
assert questions._validate_new_tag("new_tag") is True
|
||||
|
||||
|
||||
def test_validate_existing_inline_tag() -> None:
|
||||
"""Test existing tag validation."""
|
||||
questions = Questions(vault=VAULT)
|
||||
assert "Tag cannot be empty" in questions._validate_existing_inline_tag("")
|
||||
assert "'test' does not exist" in questions._validate_existing_inline_tag("test")
|
||||
assert questions._validate_existing_inline_tag("shared_tag") is True
|
||||
|
||||
|
||||
def test_validate_key_exists_regex() -> None:
|
||||
"""Test key exists regex validation."""
|
||||
questions = Questions(vault=VAULT)
|
||||
assert "'test' does not exist" in questions._validate_key_exists_regex("test")
|
||||
assert "Key cannot be empty" in questions._validate_key_exists_regex("")
|
||||
assert "Invalid regex" in questions._validate_key_exists_regex("[")
|
||||
assert questions._validate_key_exists_regex(r"\w+_Key\d") is True
|
||||
|
||||
|
||||
def test_validate_value() -> None:
|
||||
"""Test value validation."""
|
||||
questions = Questions(vault=VAULT)
|
||||
assert questions._validate_value("test") is True
|
||||
assert "Value cannot be empty" in questions._validate_value("")
|
||||
|
||||
questions2 = Questions(vault=VAULT, key="frontmatter_Key1")
|
||||
assert questions2._validate_value("test") == "frontmatter_Key1:test does not exist"
|
||||
assert "Value cannot be empty" in questions2._validate_value("")
|
||||
assert questions2._validate_value("author name") is True
|
||||
|
||||
|
||||
def test_validate_value_exists_regex() -> None:
|
||||
"""Test value exists regex validation."""
|
||||
questions2 = Questions(vault=VAULT, key="frontmatter_Key1")
|
||||
assert "Invalid regex" in questions2._validate_value_exists_regex("[")
|
||||
assert "Regex cannot be empty" in questions2._validate_value_exists_regex("")
|
||||
assert (
|
||||
questions2._validate_value_exists_regex(r"\d\d\d\w\d")
|
||||
== r"No values in frontmatter_Key1 match regex: \d\d\d\w\d"
|
||||
)
|
||||
assert questions2._validate_value_exists_regex(r"^author \w+") is True
|
||||
|
||||
|
||||
def test_validate_new_value() -> None:
|
||||
"""Test new value validation."""
|
||||
questions = Questions(vault=VAULT, key="frontmatter_Key1")
|
||||
assert questions._validate_new_value("new_value") is True
|
||||
assert "Value cannot be empty" in questions._validate_new_value("")
|
||||
assert (
|
||||
questions._validate_new_value("author name")
|
||||
== "frontmatter_Key1:author name already exists"
|
||||
)
|
||||
|
||||
@@ -190,8 +190,8 @@ def test_delete_inline_tag(test_vault) -> None:
|
||||
vault_config = config.vaults[0]
|
||||
vault = Vault(config=vault_config)
|
||||
|
||||
assert vault.delete_inline_tag("no tag") is False
|
||||
assert vault.delete_inline_tag("intext_tag2") is True
|
||||
assert vault.delete_inline_tag("no tag") == 0
|
||||
assert vault.delete_inline_tag("intext_tag2") == 2
|
||||
assert vault.metadata.dict["Inline Tags"] == [
|
||||
"ignored_file_tag2",
|
||||
"inline_tag_bottom1",
|
||||
@@ -227,8 +227,8 @@ def test_rename_inline_tag(test_vault) -> None:
|
||||
vault_config = config.vaults[0]
|
||||
vault = Vault(config=vault_config)
|
||||
|
||||
assert vault.rename_inline_tag("no tag", "new_tag") is False
|
||||
assert vault.rename_inline_tag("intext_tag2", "new_tag") is True
|
||||
assert vault.rename_inline_tag("no tag", "new_tag") == 0
|
||||
assert vault.rename_inline_tag("intext_tag2", "new_tag") == 2
|
||||
assert vault.metadata.dict["Inline Tags"] == [
|
||||
"ignored_file_tag2",
|
||||
"inline_tag_bottom1",
|
||||
@@ -248,10 +248,10 @@ def test_rename_metadata(test_vault) -> None:
|
||||
vault_config = config.vaults[0]
|
||||
vault = Vault(config=vault_config)
|
||||
|
||||
assert vault.rename_metadata("no key", "new_key") is False
|
||||
assert vault.rename_metadata("tags", "nonexistent_value", "new_vaule") is False
|
||||
assert vault.rename_metadata("no key", "new_key") == 0
|
||||
assert vault.rename_metadata("tags", "nonexistent_value", "new_vaule") == 0
|
||||
|
||||
assert vault.rename_metadata("tags", "frontmatter_tag1", "new_vaule") is True
|
||||
assert vault.rename_metadata("tags", "frontmatter_tag1", "new_vaule") == 2
|
||||
assert vault.metadata.dict["tags"] == [
|
||||
"frontmatter_tag2",
|
||||
"frontmatter_tag3",
|
||||
@@ -261,7 +261,7 @@ def test_rename_metadata(test_vault) -> None:
|
||||
"📅/frontmatter_tag3",
|
||||
]
|
||||
|
||||
assert vault.rename_metadata("tags", "new_key") is True
|
||||
assert vault.rename_metadata("tags", "new_key") == 2
|
||||
assert "tags" not in vault.metadata.dict
|
||||
assert vault.metadata.dict["new_key"] == [
|
||||
"frontmatter_tag2",
|
||||
|
||||
Reference in New Issue
Block a user