diff --git a/pytransifex/api.py b/pytransifex/api.py index 14aa9cb..05557a9 100644 --- a/pytransifex/api.py +++ b/pytransifex/api.py @@ -1,7 +1,6 @@ import logging -from os import mkdir from pathlib import Path -from typing import Any +from typing import Any, Optional import requests from transifex.api import transifex_api as tx_api @@ -18,9 +17,11 @@ class Client(Tx): """ The proper Transifex client expected by the cli and other consumers. By default instances are created and logged in 'lazyly' -- when creation or login cannot be deferred any longer. + Methods with more than 2 parameters don't allow for positional arguments. + '**kwargs' is used in methods that may need to forward extra named arguments to the API. """ - def __init__(self, config: ApiConfig, defer_login: bool = False): + def __init__(self, config: ApiConfig, defer_login=False, reset=False): """Extract config values, consumes API token against SDK client""" self.api_token = config.api_token self.host = config.host_name @@ -48,44 +49,28 @@ def login(self): @ensure_login def create_project( self, + *, project_slug: str, project_name: str | None = None, source_language_code: str = "en_GB", private: bool = False, - *args, # absorbing extra args - **kwargs, # absorbing extra kwargs + **kwargs, ) -> None | Resource: """Create a project.""" source_language = tx_api.Language.get(code=source_language_code) - - if project_name is None: - project_name = project_slug + project_name = project_name or project_slug try: - if private: - return tx_api.Project.create( - name=project_name, - slug=project_slug, - source_language=source_language, - private=private, - organization=self.organization, - ) - else: - if repository_url := kwargs.get("repository_url", None): - return tx_api.Project.create( - name=project_name, - slug=project_slug, - source_language=source_language, - private=private, - repository_url=repository_url, - organization=self.organization, - ) - else: - raise ValueError( - f"Private projects need to pass a 'repository_url' (non-empty string) argument." - ) - - logging.info("Project created!") + proj = tx_api.Project.create( + name=project_name, + slug=project_slug, + source_language=source_language, + private=private, + organization=self.organization, + **kwargs, + ) + logging.info(f"Project created with name '{project_name}' !") + return proj except JsonApiException as error: if hasattr(error, "detail") and "already exists" in error.detail: # type: ignore @@ -113,21 +98,23 @@ def list_resources(self, project_slug: str) -> list[Any]: else: return [] - raise Exception( + raise ValueError( f"Unable to find any project under this organization: '{self.organization}'" ) @ensure_login def create_resource( self, + *, project_slug: str, - path_to_file: Path, + path_to_file: str, resource_slug: str | None = None, resource_name: str | None = None, + **kwargs, ): """Create a resource using the given file contents, slugs and names""" if not (resource_slug or resource_name): - raise Exception("Please give either a resource_slug or a resource_name") + raise ValueError("Please give either a resource_slug or a resource_name") if project := self.get_project(project_slug=project_slug): resource = tx_api.Resource.create( @@ -135,6 +122,7 @@ def create_resource( name=resource_name or resource_slug, slug=resource_slug or resource_name, i18n_format=tx_api.I18nFormat(id=self.i18n_type), + **kwargs, ) with open(path_to_file, "r") as fh: @@ -144,20 +132,20 @@ def create_resource( logging.info(f"Resource created: {resource_slug or resource_name}") else: - raise Exception( + raise ValueError( f"Not project could be found wiht the slug '{project_slug}'. Please create a project first." ) @ensure_login def update_source_translation( - self, project_slug: str, resource_slug: str, path_to_file: Path + self, project_slug: str, resource_slug: str, path_to_file: str ): """ Update the translation strings for the given resource using the content of the file passsed as argument """ if not "slug" in self.organization.attributes: - raise Exception( + raise ValueError( "Unable to fetch resource for this organization; define an 'organization slug' first." ) @@ -175,7 +163,7 @@ def update_source_translation( logging.info(f"Source updated for resource: {resource_slug}") return - raise Exception( + raise ValueError( f"Unable to find resource '{resource_slug}' in project '{project_slug}'" ) @@ -185,11 +173,24 @@ def get_translation( project_slug: str, resource_slug: str, language_code: str, - output_dir: Path, - ): + path_to_output_file: None | str = None, + path_to_output_dir: None | str = None, + ) -> str: """Fetch the translation resource matching the given language""" + if path_to_output_dir and not path_to_output_file: + path_to_parent = Path(path_to_output_dir) + path_to_output_file = str( + path_to_parent.joinpath(f"{resource_slug}_{language_code}") + ) + elif path_to_output_file and not path_to_output_dir: + path_to_parent = Path(path_to_output_file).parent + else: + raise ValueError( + f"get_translation needs exactly one between 'path_to_output_file' (str) or 'path_to_output_dir (str)'. " + ) + + Path.mkdir(path_to_parent, parents=True, exist_ok=True) language = tx_api.Language.get(code=language_code) - file_name = Path.joinpath(output_dir, resource_slug) if project := self.get_project(project_slug=project_slug): if resources := project.fetch("resources"): @@ -198,54 +199,69 @@ def get_translation( resource=resource, language=language ) translated_content = requests.get(url).text - - if not Path.exists(output_dir): - mkdir(output_dir) - - with open(file_name, "w") as fh: + with open(path_to_output_file, "w") as fh: fh.write(translated_content) logging.info( f"Translations downloaded and written to file (resource: {resource_slug})" ) + return str(path_to_output_file) else: - raise Exception( + raise ValueError( f"Unable to find any resource with this slug: '{resource_slug}'" ) else: - raise Exception( + raise ValueError( f"Unable to find any resource for this project: '{project_slug}'" ) else: - raise Exception( + raise ValueError( f"Couldn't find any project with this slug: '{project_slug}'" ) @ensure_login - def list_languages(self, project_slug: str) -> list[Any]: + def list_languages(self, project_slug: str, resource_slug: str) -> list[str]: """ - List all languages for which there is at least 1 resource registered - under the parameterised project + List languages for which there exist translations under the given resource. """ if self.projects: if project := self.projects.get(slug=project_slug): - languages = project.fetch("languages") - logging.info(f"Obtained these languages") - return languages - raise Exception( + if resource := project.fetch("resources").get(slug=resource_slug): + it = tx_api.ResourceLanguageStats.filter( + project=project, resource=resource + ).all() + + language_codes = [] + for tr in it: + """ + FIXME + This is hideous and probably unsound for some language_codes. + Couldn't find a more direct accessor to language codes. + """ + code = str(tr).rsplit("_", 1)[-1][:-1] + language_codes.append(code) + + logging.info(f"Obtained these languages: {language_codes}") + return language_codes + + raise ValueError( + f"Unable to find any resource with this slug: '{resource_slug}'" + ) + raise ValueError( f"Unable to find any project with this slug: '{project_slug}'" ) - raise Exception( + raise ValueError( f"Unable to find any project under this organization: '{self.organization}'" ) @ensure_login def create_language( self, + *, project_slug: str, language_code: str, - coordinators: None | list[Any] = None, + coordinators: None | list[str] = None, ): """Create a new language resource in the remote Transifex repository""" if project := self.get_project(project_slug=project_slug): @@ -254,14 +270,22 @@ def create_language( if coordinators: project.add("coordinators", coordinators) - logging.info(f"Created language resource for {language_code}") + logging.info( + f"Created language resource for {language_code} and added these coordinators: {coordinators}" + ) @ensure_login def project_exists(self, project_slug: str) -> bool: """Check if the project exists in the remote Transifex repository""" - if self.projects and self.projects.get(slug=project_slug): - return True - return False + try: + if not self.projects: + return False + elif self.projects.get(slug=project_slug): + return True + else: + return False + except DoesNotExist: + return False @ensure_login def ping(self) -> bool: @@ -279,21 +303,22 @@ def get_project_stats(self, project_slug: str) -> dict[str, Any]: if resource_stats := tx_api.ResourceLanguageStats(project=project): return resource_stats.to_dict() - raise Exception(f"Unable to find translation for this project {project_slug}") + raise ValueError(f"Unable to find translation for this project {project_slug}") @ensure_login def pull( self, + *, project_slug: str, resource_slugs: list[str], language_codes: list[str], - output_dir: Path, + path_to_output_dir: str, ): """Pull resources from project.""" args = [] for l_code in language_codes: for slug in resource_slugs: - args.append(tuple([project_slug, slug, l_code, output_dir])) + args.append(tuple([project_slug, slug, l_code, path_to_output_dir])) res = concurrently( fn=self.get_translation, @@ -304,12 +329,12 @@ def pull( @ensure_login def push( - self, project_slug: str, resource_slugs: list[str], path_to_files: list[Path] + self, *, project_slug: str, resource_slugs: list[str], path_to_files: list[str] ): """Push resources with files under project.""" if len(resource_slugs) != len(path_to_files): raise ValueError( - f"Resources slugs ({len(resource_slugs)}) and Path to files ({len(path_to_files)}) must be equal in size!" + f"Resources slugs ({len(resource_slugs)}) and path to files ({len(path_to_files)}) must be equal in size!" ) resource_zipped_with_path = list(zip(resource_slugs, path_to_files)) @@ -352,7 +377,7 @@ class Transifex: client = None - def __new__(cls, *, defer_login: bool = False, **kwargs): + def __new__(cls, *, defer_login: bool = False, **kwargs) -> Optional["Client"]: if not cls.client: try: if kwargs: @@ -363,11 +388,9 @@ def __new__(cls, *, defer_login: bool = False, **kwargs): ) config = ApiConfig.from_env() - cls.client = Client(config, defer_login) + return Client(config, defer_login) - except Exception as error: + except ValueError as error: available = list(ApiConfig._fields) msg = f"Unable to define a proper config. API initialization uses the following fields, with only 'project_slug' optional: {available}" logging.error(f"{msg}:\n{error}") - - return cls.client diff --git a/pytransifex/cli.py b/pytransifex/cli.py index ca64746..1900d77 100644 --- a/pytransifex/cli.py +++ b/pytransifex/cli.py @@ -9,6 +9,7 @@ from pytransifex.config import CliSettings client = Transifex(defer_login=True) +assert client def path_to_slug(file_paths: list[Path]) -> list[str]: @@ -94,7 +95,7 @@ def push(input_directory: str | None): client.push( project_slug=settings.project_slug, resource_slugs=slugs, - path_to_files=files, + path_to_files=[str(f) for f in files], ) except Exception as error: reply += f"cli:push > Failed because of this error: {error}" @@ -107,16 +108,15 @@ def push(input_directory: str | None): @click.option("-l", "--only-lang", default="all") @click.option("-out", "--output-directory", is_flag=False) @cli.command("pull", help="Pull translation strings") -def pull(output_directory: str | Path | None, only_lang: str | None): +def pull(output_directory: str | None, only_lang: str | None): reply = "" settings = CliSettings.from_disk() language_codes = only_lang.split(",") if only_lang else [] if output_directory: - output_directory = Path(output_directory) - settings.output_directory = output_directory + settings.output_directory = Path(output_directory) else: - output_directory = settings.output_directory + output_directory = str(settings.output_directory) resource_slugs = [] try: @@ -127,7 +127,7 @@ def pull(output_directory: str | Path | None, only_lang: str | None): project_slug=settings.project_slug, resource_slugs=resource_slugs, language_codes=language_codes, - output_dir=output_directory, + path_to_output_dir=output_directory, ) except Exception as error: reply += f"cli:pull > failed because of this error: {error}" diff --git a/pytransifex/config.py b/pytransifex/config.py index 93d8f87..18131ec 100644 --- a/pytransifex/config.py +++ b/pytransifex/config.py @@ -22,9 +22,18 @@ def from_env(cls) -> "ApiConfig": organization = environ.get("ORGANIZATION") i18n_type = environ.get("I18N_TYPE", "PO") - if any(not v for v in [token, organization, i18n_type]): - raise Exception( - "Envars 'TX_TOKEN' and 'ORGANIZATION' must be set to non-empty values. Aborting now." + if faulty := next( + filter( + lambda v: not v[1], + zip( + ["token", "organization", "i18_ntype"], + [token, organization, i18n_type], + ), + ), + None, + ): + raise ValueError( + f"Envars 'TX_TOKEN', 'ORGANIZATION' and 'I18N_TYPE must be set to non-empty values, yet this one was found missing ('None' or empty string): {faulty[0]}" ) return cls(token, organization, i18n_type) # type: ignore diff --git a/tests/__init__.py b/tests/__init__.py index e69de29..44bf1b2 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -0,0 +1,16 @@ +import logging +from pathlib import Path + +import toml + +logging.basicConfig(level=logging.INFO) + +private = Path.cwd().joinpath("./tests/data/test_config.toml") +test_config = toml.load(private) + +public = Path.cwd().joinpath("./tests/data/test_config_public.toml") +test_config_public = toml.load(public) + +logging.info( + f"Running tests with this test_config: {test_config} and test_config_public: {test_config_public}" +) diff --git a/tests/input/test_resource_fr.po b/tests/data/resources/test_resource_fr.po similarity index 100% rename from tests/input/test_resource_fr.po rename to tests/data/resources/test_resource_fr.po diff --git a/tests/data/test_config.toml b/tests/data/test_config.toml new file mode 100644 index 0000000..56f6796 --- /dev/null +++ b/tests/data/test_config.toml @@ -0,0 +1,5 @@ +organization_slug = "test_pytransifex" +project_slug = "test_project_pytransifex" +project_name = "Python Transifex API testing" +resource_slug = "test_resource_fr" +resource_name = "Test Resource FR" diff --git a/tests/data/test_config_public.toml b/tests/data/test_config_public.toml new file mode 100644 index 0000000..b3d5fd7 --- /dev/null +++ b/tests/data/test_config_public.toml @@ -0,0 +1,6 @@ +organization_slug = "test_pytransifex" +project_slug = "test_project_pytransifex_public" +project_name = "Python Transifex API testing (public)" +repository_url = "https://github.com/opengisch/pytransifex" +resource_slug = "test_resource_fr" +resource_name = "Test Resource FR" diff --git a/tests/test_api.py b/tests/test_api.py index 25efa03..ce9a24e 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -4,18 +4,25 @@ from pytransifex.api import Transifex from pytransifex.interfaces import Tx +from tests import logging, test_config_public class TestNewApi(unittest.TestCase): @classmethod def setUpClass(cls): - cls.tx = Transifex(defer_login=True) - cls.project_slug = "test_project_pytransifex" - cls.project_name = "Test Project PyTransifex" - cls.resource_slug = "test_resource_fr" - cls.resource_name = "Test Resource FR" - cls.path_to_file = Path.cwd().joinpath("tests", "input", "test_resource_fr.po") + client = Transifex(defer_login=True) + assert client + + cls.path_to_input_dir = Path.cwd().joinpath("tests", "data", "resources") + cls.path_to_file = cls.path_to_input_dir.joinpath("test_resource_fr.po") cls.output_dir = Path.cwd().joinpath("tests", "output") + cls.output_file = cls.output_dir.joinpath("test_resource_fr_DOWNLOADED.po") + + cls.tx = client + cls.project_slug = test_config_public["project_slug"] + cls.project_name = test_config_public["project_name"] + cls.resource_slug = test_config_public["resource_slug"] + cls.resource_name = test_config_public["resource_name"] if missing := next(filter(lambda p: not p.exists(), [cls.path_to_file]), None): raise ValueError( @@ -23,10 +30,10 @@ def setUpClass(cls): ) if project := cls.tx.get_project(project_slug=cls.project_slug): - print("Found old project, removing.") + logging.info("Found old project, removing.") project.delete() - print("Creating a brand new project") + logging.info("Creating a brand new project") cls.tx.create_project( project_name=cls.project_name, project_slug=cls.project_slug, private=True ) @@ -43,46 +50,64 @@ def test2_create_project(self): # Done in setUpClass pass - def test3_list_resources(self): - _ = self.tx.list_resources(project_slug=self.project_slug) - assert True - - def test4_create_resource(self): + def test3_create_resource(self): self.tx.create_resource( project_slug=self.project_slug, resource_name=self.resource_name, resource_slug=self.resource_slug, - path_to_file=self.path_to_file, + path_to_file=str(self.path_to_file), ) assert True + def test4_list_resources(self): + resources = self.tx.list_resources(project_slug=self.project_slug) + logging.info(f"Resources found: {resources}") + assert resources + def test5_update_source_translation(self): self.tx.update_source_translation( project_slug=self.project_slug, - path_to_file=self.path_to_file, + path_to_file=str(self.path_to_file), resource_slug=self.resource_slug, ) assert True def test6_create_language(self): - self.tx.create_language(self.project_slug, "fr_CH") + self.tx.create_language(project_slug=self.project_slug, language_code="fr_CH") def test7_list_languages(self): - languages = self.tx.list_languages(project_slug=self.project_slug) - assert languages is not None + languages = self.tx.list_languages( + project_slug=self.project_slug, resource_slug=self.resource_slug + ) + logging.info(f"Languages found: {languages}") + assert languages def test8_get_translation(self): - self.tx.get_translation( + path_to_output_file = self.tx.get_translation( project_slug=self.project_slug, resource_slug=self.resource_slug, language_code="fr_CH", - output_dir=self.output_dir, + path_to_output_file=str(self.output_file), ) - assert Path.exists(Path.joinpath(self.output_dir, self.resource_slug)) + assert Path.exists(Path(path_to_output_file)) + + from difflib import Differ + from filecmp import cmp + from sys import stdout + + diff = Differ() + if not cmp(path_to_output_file, self.path_to_file): + with open(path_to_output_file, "r") as fh: + f1 = fh.readlines() + with open(path_to_output_file, "r") as fh: + f2 = fh.readlines() + res = list(diff.compare(f1, f2)) + logging.warning(f"Notice that the two files were found to differ:") + stdout.writelines(res) def test9_project_exists(self): verdict = self.tx.project_exists(project_slug=self.project_slug) - assert verdict is not None + assert verdict def test10_ping(self): self.tx.ping() @@ -90,7 +115,7 @@ def test10_ping(self): def test11_stats(self): stats = self.tx.get_project_stats(project_slug=self.project_slug) - print(str(stats)) + logging.info(str(stats)) assert stats def test12_stats(self): diff --git a/tests/test_cli.py b/tests/test_cli.py index 3b8af21..b0a0498 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -6,20 +6,25 @@ from pytransifex.api import Transifex from pytransifex.cli import cli +from tests import logging, test_config_public class TestCli(unittest.TestCase): @classmethod def setUpClass(cls): - cls.tx = Transifex(defer_login=True) - cls.project_slug = "test_project_pytransifex" - cls.project_name = "Test Project PyTransifex" - cls.resource_slug = "test_resource_fr" - cls.resource_name = "Test Resource FR" - cls.path_to_input_dir = Path.cwd().joinpath("tests", "input") + client = Transifex(defer_login=True) + assert client + + cls.path_to_input_dir = Path.cwd().joinpath("tests", "data", "resources") cls.path_to_file = cls.path_to_input_dir.joinpath("test_resource_fr.po") cls.output_dir = Path.cwd().joinpath("tests", "output") + cls.tx = client + cls.project_slug = test_config_public["project_slug"] + cls.project_name = test_config_public["project_name"] + cls.resource_slug = test_config_public["resource_slug"] + cls.resource_name = test_config_public["resource_name"] + if missing := next( filter(lambda p: not p.exists(), [cls.path_to_file, cls.path_to_input_dir]), None, @@ -29,10 +34,10 @@ def setUpClass(cls): ) if project := cls.tx.get_project(project_slug=cls.project_slug): - print("Found old project, removing.") + logging.info("Found old project, removing.") project.delete() - print("Creating a brand new project") + logging.info("Creating a brand new project") cls.tx.create_project( project_name=cls.project_name, project_slug=cls.project_slug, private=True ) @@ -55,13 +60,13 @@ def test1_init(self): def test2_push(self): result = self.runner.invoke(cli, ["push", "-in", str(self.path_to_input_dir)]) passed = result.exit_code == 0 - print(result.output) + logging.info(result.output) assert passed def test3_pull(self): result = self.runner.invoke(cli, ["pull", "-l", "fr_CH,en_GB"]) passed = result.exit_code == 0 - print(result.output) + logging.info(result.output) assert passed diff --git a/tests/test_public_project.py b/tests/test_public_project.py new file mode 100644 index 0000000..00ec364 --- /dev/null +++ b/tests/test_public_project.py @@ -0,0 +1,57 @@ +import unittest +from os import remove +from pathlib import Path + +from pytransifex.api import Transifex +from tests import logging, test_config_public + + +class TestCli(unittest.TestCase): + @classmethod + def setUpClass(cls): + client = Transifex(defer_login=True) + assert client + + cls.path_to_input_dir = Path.cwd().joinpath("tests", "data", "resources") + cls.path_to_file = cls.path_to_input_dir.joinpath("test_resource_fr.po") + cls.output_dir = Path.cwd().joinpath("tests", "output") + + cls.tx = client + cls.project_slug = test_config_public["project_slug"] + cls.project_name = test_config_public["project_name"] + cls.resource_slug = test_config_public["resource_slug"] + cls.resource_name = test_config_public["resource_name"] + repository_url = test_config_public["repository_url"] + + if missing := next( + filter(lambda p: not p.exists(), [cls.path_to_file, cls.path_to_input_dir]), + None, + ): + raise ValueError( + f"Unable to complete test with broken tests inputs. Found missing: {missing}" + ) + + if project := cls.tx.get_project(project_slug=cls.project_slug): + logging.info("Found old project, removing.") + project.delete() + + logging.info("Creating a brand new project") + cls.tx.create_project( + project_name=cls.project_name, + project_slug=cls.project_slug, + private=False, + repository_url=repository_url, + ) + + @classmethod + def tearDownClass(cls): + if Path.exists(cls.output_dir): + remove(cls.output_dir) + + def test1_project_exists(self): + verdict = self.tx.project_exists(project_slug=self.project_slug) + assert verdict + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_utils.py b/tests/test_utils.py index b5d4b35..b9edcb5 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -26,3 +26,7 @@ def test1_map_async(self): def test2_map_async(self): res = concurrently(fn=fn, args=self.args) assert res == self.res + + +if __name__ == "__main__": + unittest.main()