# Path: internetarchive/utils.py
# def deep_update(d, u):
# def validate_s3_identifier(string):
# def needs_quote(s):
# def norm_filepath(fp):
# def get_md5(file_object):
# def chunk_generator(fp, chunk_size):
# def suppress_keyboard_interrupt_message():
#     def new_hook(type, value, traceback):
#     def __init__(self, iterable, size):
#     def read(self, size=-1):  # TBD: add buffer for `len(data) > size` case
#     def __len__(self):
#     def __init__(self, id_list_or_single_id, session):
#     def __len__(self):
#     def __getitem__(self, idx):
#     def __getattr__(self, name):
#     def __repr__(self):
# def get_s3_xml_text(xml_str):
#     def _get_tag_text(tag_name, xml_obj):
# def get_file_size(file_obj):
# def iter_directory(directory):
# def recursive_file_count(files, item=None, checksum=False):
# def is_dir(obj):
# def reraise_modify(caught_exc, append_msg, prepend=False):
# def remove_none(obj):
# def delete_items_from_dict(d, to_delete):
# def is_valid_metadata_key(name):
# def merge_dictionaries(dict0, dict1, keys_to_drop=None):
# def parse_dict_cookies(value):
# class InvalidIdentifierException(Exception):
# class IterableToFileAdapter:
# class IdentifierListAsItems:
# 
# Path: tests/conftest.py
# class IaRequestsMock(RequestsMock):
#     def add_metadata_mock(self, identifier, body=None, method=responses.GET,
#                           protocol='https?', transform_body=None):
#         url = re.compile(f'{protocol}://archive.org/metadata/{identifier}')
#         if body is None:
#             body = load_test_data_file(f'metadata/{identifier}.json')
#         if transform_body:
#             body = transform_body(body)
#         self.add(method, url, body=body, content_type='application/json')
# 
#     def mock_all_downloads(self, num_calls=1, body='test content', protocol='https?'):
#         url = re.compile(f'{protocol}://archive.org/download/.*')
#         for _ in range(6):
#             self.add(responses.GET, url, body=body)
# 
# def load_test_data_file(filename):
#     return load_file(os.path.join(ROOT_DIR, 'tests/data/', filename))
# 
# def ia_call(argv, expected_exit_code=0):
#     # Use a test config for all `ia` tests.
#     argv.insert(1, '--config-file')
#     argv.insert(2, TEST_CONFIG)
#     sys.argv = argv
#     try:
#         ia.main()
#     except SystemExit as exc:
#         exit_code = exc.code if exc.code else 0
#         assert exit_code == expected_exit_code


import responses
import os
import sys
from contextlib import contextmanager
from io import StringIO
from internetarchive.utils import json
from tests.conftest import IaRequestsMock, load_test_data_file, ia_call

    # Default behaviour
    with IaRequestsMock() as rsps:
        rsps.add_metadata_mock('nasa')
        rsps.add(responses.PUT, f'{PROTOCOL}//s3.us.archive.org/nasa/test.txt',
                 body='',
                 content_type='text/plain')
        ia_call(['ia', '--log', 'upload', 'nasa', 'foo/test.txt'])
    assert f'uploaded test.txt to {PROTOCOL}//s3.us.archive.org/nasa/test.txt' in caplog.text
    caplog.clear()

    with IaRequestsMock() as rsps:
        rsps.add_metadata_mock('nasa')
        rsps.add(responses.PUT, f'{PROTOCOL}//s3.us.archive.org/nasa/test.txt',
                 body='',
                 content_type='text/plain')
        ia_call(['ia', '--log', 'upload', '--spreadsheet', 'test.csv'])
    assert f'uploaded test.txt to {PROTOCOL}//s3.us.archive.org/nasa/test.txt' in caplog.text
    caplog.clear()

    # With the option
    with IaRequestsMock() as rsps:
        rsps.add_metadata_mock('nasa')
        rsps.add(responses.PUT, f'{PROTOCOL}//s3.us.archive.org/nasa/foo/test.txt',
                 body='',
                 content_type='text/plain')
        ia_call(['ia', '--log', 'upload', 'nasa', 'foo/test.txt', '--keep-directories'])
    assert f'uploaded foo/test.txt to {PROTOCOL}//s3.us.archive.org/nasa/foo/test.txt' in caplog.text
    caplog.clear()


# Path: internetarchive/utils.py
# def deep_update(d, u):
# def validate_s3_identifier(string):
# def needs_quote(s):
# def norm_filepath(fp):
# def get_md5(file_object):
# def chunk_generator(fp, chunk_size):
# def suppress_keyboard_interrupt_message():
#     def new_hook(type, value, traceback):
#     def __init__(self, iterable, size):
#     def read(self, size=-1):  # TBD: add buffer for `len(data) > size` case
#     def __len__(self):
#     def __init__(self, id_list_or_single_id, session):
#     def __len__(self):
#     def __getitem__(self, idx):
#     def __getattr__(self, name):
#     def __repr__(self):
# def get_s3_xml_text(xml_str):
#     def _get_tag_text(tag_name, xml_obj):
# def get_file_size(file_obj):
# def iter_directory(directory):
# def recursive_file_count(files, item=None, checksum=False):
# def is_dir(obj):
# def reraise_modify(caught_exc, append_msg, prepend=False):
# def remove_none(obj):
# def delete_items_from_dict(d, to_delete):
# def is_valid_metadata_key(name):
# def merge_dictionaries(dict0, dict1, keys_to_drop=None):
# def parse_dict_cookies(value):
# class InvalidIdentifierException(Exception):
# class IterableToFileAdapter:
# class IdentifierListAsItems:
# 
# Path: tests/conftest.py
# class IaRequestsMock(RequestsMock):
#     def add_metadata_mock(self, identifier, body=None, method=responses.GET,
#                           protocol='https?', transform_body=None):
#         url = re.compile(f'{protocol}://archive.org/metadata/{identifier}')
#         if body is None:
#             body = load_test_data_file(f'metadata/{identifier}.json')
#         if transform_body:
#             body = transform_body(body)
#         self.add(method, url, body=body, content_type='application/json')
# 
#     def mock_all_downloads(self, num_calls=1, body='test content', protocol='https?'):
#         url = re.compile(f'{protocol}://archive.org/download/.*')
#         for _ in range(6):
#             self.add(responses.GET, url, body=body)
# 
# def load_test_data_file(filename):
#     return load_file(os.path.join(ROOT_DIR, 'tests/data/', filename))
# 
# def ia_call(argv, expected_exit_code=0):
#     # Use a test config for all `ia` tests.
#     argv.insert(1, '--config-file')
#     argv.insert(2, TEST_CONFIG)
#     sys.argv = argv
#     try:
#         ia.main()
#     except SystemExit as exc:
#         exit_code = exc.code if exc.code else 0
#         assert exit_code == expected_exit_code


# Path: tests/cli/test_ia_upload.py
import responses
import os
import sys
from contextlib import contextmanager
from io import StringIO
from internetarchive.utils import json
from tests.conftest import IaRequestsMock, load_test_data_file, ia_call


    # Default behaviour
    with IaRequestsMock() as rsps:
        rsps.add_metadata_mock('nasa')
        rsps.add(responses.PUT, f'{PROTOCOL}//s3.us.archive.org/nasa/test.txt',
                 body='',
                 content_type='text/plain')
        ia_call(['ia', '--log', 'upload', 'nasa', 'foo/test.txt'])
    assert f'uploaded test.txt to {PROTOCOL}//s3.us.archive.org/nasa/test.txt' in caplog.text
    caplog.clear()

    with IaRequestsMock() as rsps:
        rsps.add_metadata_mock('nasa')
        rsps.add(responses.PUT, f'{PROTOCOL}//s3.us.archive.org/nasa/test.txt',
                 body='',
                 content_type='text/plain')
        ia_call(['ia', '--log', 'upload', '--spreadsheet', 'test.csv'])
    assert f'uploaded test.txt to {PROTOCOL}//s3.us.archive.org/nasa/test.txt' in caplog.text
    caplog.clear()

    # With the option
    with IaRequestsMock() as rsps:
        rsps.add_metadata_mock('nasa')
        rsps.add(responses.PUT, f'{PROTOCOL}//s3.us.archive.org/nasa/foo/test.txt',
                 body='',
                 content_type='text/plain')
        ia_call(['ia', '--log', 'upload', 'nasa', 'foo/test.txt', '--keep-directories'])
    assert f'uploaded foo/test.txt to {PROTOCOL}//s3.us.archive.org/nasa/foo/test.txt' in caplog.text
    caplog.clear()


    with IaRequestsMock() as rsps: