212 lines
6.6 KiB
Python
212 lines
6.6 KiB
Python
# coding=utf-8
|
|
from unittest import TestCase
|
|
import os
|
|
import shutil
|
|
from vcr import VCR
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
from icloudpd.base import main
|
|
import inspect
|
|
import glob
|
|
|
|
from tests.helpers import path_from_project_root, print_result_exception, recreate_path
|
|
|
|
vcr = VCR(decode_compressed_response=True)
|
|
|
|
|
|
class CliTestCase(TestCase):
|
|
@pytest.fixture(autouse=True)
|
|
def inject_fixtures(self, caplog):
|
|
self._caplog = caplog
|
|
self.root_path = path_from_project_root(__file__)
|
|
self.fixtures_path = os.path.join(self.root_path, "fixtures")
|
|
self.vcr_path = os.path.join(self.root_path, "vcr_cassettes")
|
|
|
|
def test_cli(self):
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["--help"])
|
|
assert result.exit_code == 0
|
|
|
|
def test_log_levels(self):
|
|
base_dir = os.path.join(self.fixtures_path, inspect.stack()[0][3])
|
|
cookie_dir = os.path.join(base_dir, "cookie")
|
|
data_dir = os.path.join(base_dir, "data")
|
|
|
|
for dir in [base_dir, cookie_dir, data_dir]:
|
|
recreate_path(dir)
|
|
|
|
parameters = [
|
|
("debug", ["DEBUG", "INFO"], []),
|
|
("info", ["INFO"], ["DEBUG"]),
|
|
("error", [], ["DEBUG", "INFO"]),
|
|
]
|
|
for log_level, expected, not_expected in parameters:
|
|
self._caplog.clear()
|
|
recreate_path(cookie_dir)
|
|
with vcr.use_cassette(os.path.join(self.vcr_path, "listing_photos.yml")):
|
|
# Pass fixed client ID via environment variable
|
|
runner = CliRunner(env={
|
|
"CLIENT_ID": "DE309E26-942E-11E8-92F5-14109FE0B321"
|
|
})
|
|
result = runner.invoke(
|
|
main,
|
|
[
|
|
"--username",
|
|
"jdoe@gmail.com",
|
|
"--password",
|
|
"password1",
|
|
"--recent",
|
|
"0",
|
|
"--log-level",
|
|
log_level,
|
|
"-d",
|
|
data_dir,
|
|
"--cookie-directory",
|
|
cookie_dir,
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
for text in expected:
|
|
self.assertIn(text, self._caplog.text)
|
|
for text in not_expected:
|
|
self.assertNotIn(text, self._caplog.text)
|
|
|
|
files_in_result = glob.glob(os.path.join(data_dir, "**/*.*"), recursive=True)
|
|
|
|
assert sum(1 for _ in files_in_result) == 0
|
|
|
|
def test_tqdm(self):
|
|
base_dir = os.path.join(self.fixtures_path, inspect.stack()[0][3])
|
|
cookie_dir = os.path.join(base_dir, "cookie")
|
|
data_dir = os.path.join(base_dir, "data")
|
|
|
|
for dir in [base_dir, cookie_dir, data_dir]:
|
|
recreate_path(dir)
|
|
|
|
with vcr.use_cassette(os.path.join(self.vcr_path, "listing_photos.yml")):
|
|
# Force tqdm progress bar via ENV var
|
|
runner = CliRunner(env={
|
|
"FORCE_TQDM": "yes",
|
|
"CLIENT_ID": "DE309E26-942E-11E8-92F5-14109FE0B321"
|
|
})
|
|
result = runner.invoke(
|
|
main,
|
|
[
|
|
"--username",
|
|
"jdoe@gmail.com",
|
|
"--password",
|
|
"password1",
|
|
"--recent",
|
|
"0",
|
|
"-d",
|
|
data_dir,
|
|
"--cookie-directory",
|
|
cookie_dir,
|
|
],
|
|
)
|
|
print_result_exception(result)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
files_in_result = glob.glob(os.path.join(data_dir, "**/*.*"), recursive=True)
|
|
|
|
assert sum(1 for _ in files_in_result) == 0
|
|
|
|
def test_unicode_directory(self):
|
|
base_dir = os.path.join(self.fixtures_path, inspect.stack()[0][3])
|
|
cookie_dir = os.path.join(base_dir, "cookie")
|
|
data_dir = os.path.join(base_dir, "data")
|
|
|
|
for dir in [base_dir, cookie_dir, data_dir]:
|
|
recreate_path(dir)
|
|
|
|
with vcr.use_cassette(os.path.join(self.vcr_path, "listing_photos.yml")):
|
|
# Pass fixed client ID via environment variable
|
|
runner = CliRunner(env={
|
|
"CLIENT_ID": "DE309E26-942E-11E8-92F5-14109FE0B321"
|
|
})
|
|
result = runner.invoke(
|
|
main,
|
|
[
|
|
"--username",
|
|
"jdoe@gmail.com",
|
|
"--password",
|
|
"password1",
|
|
"--recent",
|
|
"0",
|
|
"--log-level",
|
|
"info",
|
|
"-d",
|
|
data_dir,
|
|
"--cookie-directory",
|
|
cookie_dir,
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
|
|
files_in_result = glob.glob(os.path.join(data_dir, "**/*.*"), recursive=True)
|
|
|
|
assert sum(1 for _ in files_in_result) == 0
|
|
|
|
def test_missing_directory(self):
|
|
base_dir = os.path.join(self.fixtures_path, inspect.stack()[0][3])
|
|
# need path removed
|
|
if os.path.exists(base_dir):
|
|
shutil.rmtree(base_dir)
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
main,
|
|
[
|
|
"--username",
|
|
"jdoe@gmail.com",
|
|
"--password",
|
|
"password1",
|
|
"--recent",
|
|
"0",
|
|
"--log-level",
|
|
"info",
|
|
"-d",
|
|
base_dir
|
|
],
|
|
)
|
|
assert result.exit_code == 2
|
|
|
|
files_in_result = glob.glob(os.path.join(base_dir, "**/*.*"), recursive=True)
|
|
|
|
assert sum(1 for _ in files_in_result) == 0
|
|
|
|
def test_missing_directory_param(self):
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
main,
|
|
[
|
|
"--username",
|
|
"jdoe@gmail.com",
|
|
"--password",
|
|
"password1",
|
|
"--recent",
|
|
"0",
|
|
"--log-level",
|
|
"info",
|
|
],
|
|
)
|
|
assert result.exit_code == 2
|
|
|
|
def test_conflict_options_delete_after_download_and_auto_delete(self):
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
main,
|
|
[
|
|
"--username",
|
|
"jdoe@gmail.com",
|
|
"--password",
|
|
"password1",
|
|
"-d",
|
|
"/tmp",
|
|
"--delete-after-download",
|
|
"--auto-delete"
|
|
],
|
|
)
|
|
assert result.exit_code == 2
|