Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if os.path.exists("tests/fixtures/Photos"):
shutil.rmtree("tests/fixtures/Photos")
os.makedirs("tests/fixtures/Photos")
with mock.patch("icloudpd.download.download_media") as dp_patched:
dp_patched.return_value = True
with mock.patch.object(PhotoAsset, "versions") as pa:
pa.return_value = ["original", "medium"]
with vcr.use_cassette("tests/vcr_cassettes/listing_photos.yml"):
# Pass fixed client ID via environment variable
os.environ["CLIENT_ID"] = "DE309E26-942E-11E8-92F5-14109FE0B321"
runner = CliRunner()
result = runner.invoke(
main,
[
"--username",
"jdoe@gmail.com",
"--password",
"password1",
"--recent",
"1",
"--size",
"thumb",
"--force-size",
"--no-progress-bar",
"-d",
base_dir,
],
)
print_result_exception(result)
def test_2sa_flow_sms(self):
with vcr.use_cassette("tests/vcr_cassettes/2sa_flow_valid_sms.yml"):
os.environ["CLIENT_ID"] = "DE309E26-942E-11E8-92F5-14109FE0B321"
runner = CliRunner()
result = runner.invoke(
main,
[
"--username",
"jdoe@gmail.com",
"--password",
"password1",
"--recent",
"0",
"--no-progress-bar",
"-d",
"tests/fixtures/Photos",
],
input="0\n123456\n",
)
self.assertIn("DEBUG Authenticating...", self._caplog.text)
self.assertIn(
"INFO Two-step/two-factor authentication is required!",
def test_password_prompt(self):
if not os.path.exists("tests/fixtures/Photos"):
os.makedirs("tests/fixtures/Photos")
with vcr.use_cassette("tests/vcr_cassettes/listing_photos.yml"):
os.environ["CLIENT_ID"] = "DE309E26-942E-11E8-92F5-14109FE0B321"
runner = CliRunner()
result = runner.invoke(
main,
[
"--username",
"jdoe@gmail.com",
"--recent",
"0",
"--no-progress-bar",
"-d",
"tests/fixtures/Photos",
],
input="password1\n",
)
self.assertIn("DEBUG Authenticating...", self._caplog.text)
self.assertIn(
"DEBUG Looking up all photos and videos from album All Photos...", self._caplog.text
)
self.assertIn(
def test_2sa_flow_invalid_device_2fa(self):
with vcr.use_cassette("tests/vcr_cassettes/2sa_flow_invalid_device.yml"):
os.environ["CLIENT_ID"] = "DE309E26-942E-11E8-92F5-14109FE0B321"
runner = CliRunner()
result = runner.invoke(
main,
[
"--username",
"jdoe@gmail.com",
"--password",
"password1",
"--recent",
"0",
"--no-progress-bar",
"-d",
"tests/fixtures/Photos",
],
input="1\n901431\n",
)
self.assertIn(
"ERROR Failed to verify two-factor authentication code",
self._caplog.text,
def test_cli(self):
runner = CliRunner()
result = runner.invoke(main, ["--help"])
assert result.exit_code == 0
os.makedirs("tests/fixtures/Photos/2018/07/26/")
open("tests/fixtures/Photos/2018/07/26/IMG_7383.PNG", "a").close()
os.makedirs("tests/fixtures/Photos/2018/07/12/")
open("tests/fixtures/Photos/2018/07/12/IMG_7190.JPG", "a").close()
open("tests/fixtures/Photos/2018/07/12/IMG_7190-medium.JPG", "a").close()
# Should not be deleted
open("tests/fixtures/Photos/2018/07/30/IMG_7407.JPG", "a").close()
open("tests/fixtures/Photos/2018/07/30/IMG_7407-original.JPG", "a").close()
with vcr.use_cassette("tests/vcr_cassettes/autodelete_photos.yml"):
# Pass fixed client ID via environment variable
os.environ["CLIENT_ID"] = "DE309E26-942E-11E8-92F5-14109FE0B321"
runner = CliRunner()
result = runner.invoke(
main,
[
"--username",
"jdoe@gmail.com",
"--password",
"password1",
"--recent",
"0",
"--skip-videos",
"--auto-delete",
"-d",
"tests/fixtures/Photos",
],
)
self.assertIn("DEBUG Looking up all photos from album All Photos...", self._caplog.text)
self.assertIn(
"INFO Downloading 0 original photos to tests/fixtures/Photos/ ...",
def test_download_and_skip_existing_photos(self):
if os.path.exists("tests/fixtures/Photos"):
shutil.rmtree("tests/fixtures/Photos")
os.makedirs("tests/fixtures/Photos")
os.makedirs("tests/fixtures/Photos/2018/07/30/")
open("tests/fixtures/Photos/2018/07/30/IMG_7408.JPG", "a").close()
open("tests/fixtures/Photos/2018/07/30/IMG_7407.JPG", "a").close()
with vcr.use_cassette("tests/vcr_cassettes/listing_photos.yml"):
# Pass fixed client ID via environment variable
os.environ["CLIENT_ID"] = "DE309E26-942E-11E8-92F5-14109FE0B321"
runner = CliRunner()
result = runner.invoke(
main,
[
"--username",
"jdoe@gmail.com",
"--password",
"password1",
"--recent",
"5",
"--skip-videos",
"--skip-live-photos",
"--set-exif-datetime",
"--no-progress-bar",
"-d",
"tests/fixtures/Photos",
],
)
print_result_exception(result)
def test_2sa_required_email_notification(self):
with vcr.use_cassette("tests/vcr_cassettes/auth_requires_2sa.yml"):
with patch("smtplib.SMTP") as smtp:
# Pass fixed client ID via environment variable
os.environ["CLIENT_ID"] = "EC5646DE-9423-11E8-BF21-14109FE0B321"
runner = CliRunner()
result = runner.invoke(
main,
[
"--username",
"jdoe@gmail.com",
"--password",
"password1",
"--smtp-username",
"jdoe+smtp@gmail.com",
"--smtp-password",
"password1",
"--notification-email",
"jdoe+notifications@gmail.com",
"-d",
"tests/fixtures/Photos",
],
)
print(result.output)
def test_listing_albums(self):
if os.path.exists("tests/fixtures/Photos"):
shutil.rmtree("tests/fixtures/Photos")
os.makedirs("tests/fixtures/Photos")
with vcr.use_cassette("tests/vcr_cassettes/listing_albums.yml"):
# Pass fixed client ID via environment variable
os.environ["CLIENT_ID"] = "DE309E26-942E-11E8-92F5-14109FE0B321"
runner = CliRunner()
result = runner.invoke(
main,
[
"--username",
"jdoe@gmail.com",
"--password",
"password1",
"--list-albums",
"--no-progress-bar",
"-d",
"tests/fixtures/Photos",
],
)
print_result_exception(result)
albums = result.output.splitlines()
self.assertIn("All Photos", albums)
def test_log_levels(self):
if not os.path.exists("tests/fixtures/Photos"):
os.makedirs("tests/fixtures/Photos")
with vcr.use_cassette("tests/vcr_cassettes/listing_photos.yml"):
# Pass fixed client ID via environment variable
os.environ["CLIENT_ID"] = "DE309E26-942E-11E8-92F5-14109FE0B321"
runner = CliRunner()
result = runner.invoke(
main,
[
"--username",
"jdoe@gmail.com",
"--password",
"password1",
"--recent",
"0",
"--log-level",
"info",
"-d"
"tests/fixtures/Photos",
],
)
assert result.exit_code == 0
with vcr.use_cassette("tests/vcr_cassettes/listing_photos.yml"):
result = runner.invoke(