generated from bokulich-lab/q2-plugin-template
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
7 changed files
with
144 additions
and
14 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| import os | ||
| import re | ||
| import subprocess | ||
|
|
||
| from qiime2.util import duplicate | ||
|
|
||
| from q2_amr.amrfinderplus.types import AMRFinderPlusDatabaseDirFmt | ||
| from q2_amr.card.utils import run_command | ||
|
|
||
|
|
||
| def fetch_amrfinderplus_db() -> AMRFinderPlusDatabaseDirFmt: | ||
| amrfinderplus_db = AMRFinderPlusDatabaseDirFmt() | ||
|
|
||
| # Run "amrfinder -u" command that downloads the database | ||
| run_amrfinder_u() | ||
|
|
||
| # Define path where the database will be downloaded to | ||
| conda_prefix = os.getenv("CONDA_PREFIX") | ||
| amrfinder_db_path = os.path.join( | ||
| conda_prefix, "share", "amrfinderplus", "data", "latest" | ||
| ) | ||
|
|
||
| # Copy all files from amrfinder_db_path to database directory format | ||
| _copy_all(amrfinder_db_path, amrfinderplus_db.path) | ||
|
|
||
| return amrfinderplus_db | ||
|
|
||
|
|
||
| def _copy_all(src_dir, des_dir): | ||
| regex = re.compile(r".*(?:AMR_CDS|changes).*") | ||
| # Loop over all files in the source directory | ||
| for file in os.listdir(src_dir): | ||
| # Check if the filename does not match the regex pattern and copy the file | ||
| # from src to des. Files matching the pattern are not needed for the database. | ||
| if not regex.match(file): | ||
| duplicate(os.path.join(src_dir, file), os.path.join(des_dir, file)) | ||
|
|
||
|
|
||
| def run_amrfinder_u(): | ||
| # The command "amrfinder -u" downloads the latest amrfinderplus database or | ||
| # updates it | ||
| cmd = ["amrfinder", "-u"] | ||
| try: | ||
| run_command(cmd, verbose=True) | ||
| except subprocess.CalledProcessError as e: | ||
| raise Exception( | ||
| "An error was encountered while running AMRFinderPlus, " | ||
| f"(return code {e.returncode}), please inspect " | ||
| "stdout and stderr to learn more." | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,52 @@ | ||
| import os | ||
| import subprocess | ||
| from unittest.mock import patch | ||
|
|
||
| from qiime2.plugin.testing import TestPluginBase | ||
|
|
||
| from q2_amr.amrfinderplus.database import ( | ||
| _copy_all, | ||
| fetch_amrfinderplus_db, | ||
| run_amrfinder_u, | ||
| ) | ||
|
|
||
|
|
||
| class TestFetchAMRFinderPlusDB(TestPluginBase): | ||
| package = "q2_amr.amrfinderplus.tests" | ||
|
|
||
| @patch("q2_amr.amrfinderplus.database.run_amrfinder_u") | ||
| @patch("q2_amr.amrfinderplus.database._copy_all") | ||
| def test_fetch_amrfinderplus_db(self, mock_run_amrfinder_u, mock__copy_all): | ||
| fetch_amrfinderplus_db() | ||
|
|
||
| @patch("q2_amr.amrfinderplus.database.run_command") | ||
| def test_run_amrfinder_u(self, mock_run_command): | ||
| run_amrfinder_u() | ||
| mock_run_command.assert_called_once_with( | ||
| ["amrfinder", "-u"], | ||
| verbose=True, | ||
| ) | ||
|
|
||
| @patch("q2_amr.amrfinderplus.database.run_command") | ||
| def test_run_amrfinder_u_error(self, mock_run_command): | ||
| expected_message = ( | ||
| "An error was encountered while running AMRFinderPlus, " | ||
| "(return code 1), please inspect stdout and stderr to learn more." | ||
| ) | ||
| mock_run_command.side_effect = subprocess.CalledProcessError(1, "cmd") | ||
| with self.assertRaises(Exception) as cm: | ||
| run_amrfinder_u() | ||
| self.assertEqual(str(cm.exception), expected_message) | ||
|
|
||
| def test__copy_all(self): | ||
| tmp = self.temp_dir.name | ||
| os.mkdir(os.path.join(tmp, "src")) | ||
| os.mkdir(os.path.join(tmp, "des")) | ||
|
|
||
| with open(os.path.join(tmp, "src", "a"), "w"), open( | ||
| os.path.join(tmp, "src", "AMR_CDS.nto"), "w" | ||
| ): | ||
| pass | ||
|
|
||
| _copy_all(os.path.join(tmp, "src"), os.path.join(tmp, "des")) | ||
| self.assertTrue(os.path.exists(os.path.join(tmp, "des", "a"))) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters