def test_swift_authenticate_fails_with_missing_params(self, mock_create_context):
        mock_context = mock_create_context.return_value
        mock_swift = mock_context.get_client.return_value

        temp_output = tempfile.NamedTemporaryFile()
        mock_swift.fetch_object.return_value = ["FOOBAR", ]

        # uri with missing auth_endpoint... should fail.
        uri = "swift://%(username)s:%(password)[email protected]%(container)s/%(file)s?" \
              "region=%(region)s&tenant_id=%(tenant_id)s" % self.params
        storage = storagelib.get_storage(uri)
        with self.assertRaises(storagelib.storage.InvalidStorageUri):
            storage.save_to_filename(temp_output.name)

        # uri with missing region... should fail.
        uri = "swift://%(username)s:%(password)[email protected]%(container)s/%(file)s?" \
              "auth_endpoint=%(auth_endpoint)s&tenant_id=%(tenant_id)s" % self.params
        storage = storagelib.get_storage(uri)
        with self.assertRaises(storagelib.storage.InvalidStorageUri):
            storage.save_to_filename(temp_output.name)

        # uri with missing tenant_id... should fail.
        uri = "swift://%(username)s:%(password)[email protected]%(container)s/%(file)s?" \
              "auth_endpoint=%(auth_endpoint)s&region=%(region)s" % self.params
        storage = storagelib.get_storage(uri)
        with self.assertRaises(storagelib.storage.InvalidStorageUri):
            storage.save_to_filename(temp_output.name)

    def test_swift_save_to_directory(
            self, mock_timeout, mock_create_context, mock_makedirs, mock_path_exists):

        expected_jpg = self.RackspaceObject("file/a/0.jpg", "image/jpg")
        expected_mp4 = self.RackspaceObject("file/a/b/c/1.mp4", "video/mp4")

        expected_files = [
            expected_jpg,
            expected_mp4
        ]
        mock_context = mock_create_context.return_value
        mock_swift = mock_context.get_client.return_value

        mock_swift.list_container_objects.return_value = expected_files

        uri = "swift://{username}:{password}@{container}/{file}?" \
              "auth_endpoint={auth_endpoint}&region={region}" \
              "&tenant_id={tenant_id}".format(**self.params)

        storagelib.get_storage(uri).save_to_directory("/tmp/cat/pants")

        self._assert_default_login_correct(mock_create_context, mock_timeout)
        mock_swift.list_container_objects.assert_called_with(
            self.params["container"], prefix=self.params["file"])

        mock_makedirs.assert_has_calls([
            mock.call("/tmp/cat/pants/a"), mock.call("/tmp/cat/pants/a/b/c")])

        mock_swift.download_object.assert_any_call(
            self.params["container"], expected_jpg, "/tmp/cat/pants/a", structure=False)
        mock_swift.download_object.assert_any_call(
            self.params["container"], expected_mp4, "/tmp/cat/pants/a/b/c", structure=False)

def fhourstones(mark, storage='console', local_sensor=None, remote_sensor=None):
    res = do_fhourstones(local_sensor=local_sensor,
                         remote_sensor=remote_sensor)
    res['mark'] = mark
    get_storage(storage, 'fhourstones')(res)
    
    return res

    def test_swift_save_to_directory_fails_after_five_failed_file_download_retries(
            self, mock_timeout, mock_create_context, mock_makedirs, mock_path_exists, mock_uniform,
            mock_sleep):

        expected_jpg = self.RackspaceObject("file/a/0.jpg", "image/jpg")
        expected_mp4 = self.RackspaceObject("file/a/b/c/1.mp4", "video/mp4")

        expected_files = [
            expected_jpg,
            expected_mp4
        ]
        mock_context = mock_create_context.return_value
        mock_swift = mock_context.get_client.return_value

        mock_swift.list_container_objects.return_value = expected_files

        mock_swift.download_object.side_effect = [
            storagelib.storage.TimeoutError,
            IOError,
            IOError,
            IOError,
            RuntimeError,
        ]

        uri = "swift://{username}:{password}@{container}/{file}?" \
              "auth_endpoint={auth_endpoint}&region={region}" \
              "&tenant_id={tenant_id}".format(**self.params)

        with self.assertRaises(RuntimeError):
            storagelib.get_storage(uri).save_to_directory("/tmp/cat/pants")

        self._assert_default_login_correct(mock_create_context, mock_timeout)
        mock_swift.list_container_objects.assert_called_with(
            self.params["container"], prefix=self.params["file"])

        mock_makedirs.assert_called_once_with("/tmp/cat/pants/a")

        self.assertEqual(5, mock_swift.download_object.call_count)
        mock_swift.download_object.assert_has_calls([
            mock.call(self.params["container"], expected_jpg, "/tmp/cat/pants/a", structure=False),
            mock.call(self.params["container"], expected_jpg, "/tmp/cat/pants/a", structure=False),
            mock.call(self.params["container"], expected_jpg, "/tmp/cat/pants/a", structure=False),
            mock.call(self.params["container"], expected_jpg, "/tmp/cat/pants/a", structure=False),
            mock.call(self.params["container"], expected_jpg, "/tmp/cat/pants/a", structure=False)
        ])

        mock_uniform.assert_has_calls([
            mock.call(0, 1),
            mock.call(0, 3),
            mock.call(0, 7),
            mock.call(0, 15)
        ])

        self.assertEqual(4, mock_sleep.call_count)
        mock_sleep.assert_called_with(mock_uniform.return_value)

    def test_local_storage_load_from_directory(self):
        temp_directory = create_temp_nested_directory_with_files()

        temp_output_dir = tempfile.mkdtemp()
        storage = storagelib.get_storage("file://{0}/{1}".format(temp_output_dir, "tmp"))

        storage.load_from_directory(temp_directory["temp_directory"]["path"])

        destination_directory_path = os.path.join(
            temp_output_dir, "tmp")
        destination_input_one_path = os.path.join(
            temp_output_dir, destination_directory_path, temp_directory["temp_input_one"]["name"])
        destination_input_two_path = os.path.join(
            temp_output_dir, destination_directory_path, temp_directory["temp_input_two"]["name"])
        nested_temp_input_path = os.path.join(
            temp_output_dir, destination_directory_path,
            temp_directory["nested_temp_directory"]["path"],
            temp_directory["nested_temp_input"]["name"])

        self.assertTrue(os.path.exists(destination_input_one_path))
        self.assertTrue(os.path.exists(destination_input_two_path))
        self.assertTrue(os.path.exists(nested_temp_input_path))

        with open(destination_input_one_path) as temp_output_fp:
            self.assertEqual("FOO", temp_output_fp.read())

        with open(destination_input_two_path) as temp_output_fp:
            self.assertEqual("BAR", temp_output_fp.read())

        with open(nested_temp_input_path) as temp_output_fp:
            self.assertEqual("FOOBAR", temp_output_fp.read())

    def test_ftps_scheme_connects_using_ftp_tls_class(self, mock_ftp_tls_class):
        temp_output = tempfile.NamedTemporaryFile()

        mock_results = ["foo", "bar"]

        def mock_retrbinary(command, callback):
            for chunk in mock_results:
                callback(chunk)

            return "226"

        mock_ftp = mock_ftp_tls_class.return_value
        mock_ftp.retrbinary.side_effect = mock_retrbinary

        def assert_tcp_keepalive_already_enabled(username, password):
            # It is important that these already be called before
            # login is called, because FTP_TLS.login replaces the
            # socket instance with an SSL-wrapped socket.
            mock_ftp.sock.setsockopt.assert_any_call(
                socket.SOL_SOCKET, socket.SO_KEEPALIVE,
                storagelib.storage.DEFAULT_FTP_KEEPALIVE_ENABLE)

        mock_ftp.login.side_effect = assert_tcp_keepalive_already_enabled

        storage = storagelib.get_storage("ftps://user:[email protected]/some/dir/file")

        storage.save_to_filename(temp_output.name)

        mock_ftp_tls_class.assert_called_with(timeout=storagelib.storage.DEFAULT_FTP_TIMEOUT)
        mock_ftp.connect.assert_called_with("ftp.foo.com", port=21)
        mock_ftp.login.assert_called_with("user", "password")
        mock_ftp.prot_p.assert_called_with()

    def test_ftp_save_to_filename(self, mock_ftp_class):
        temp_output = tempfile.NamedTemporaryFile()

        mock_results = ["foo", "bar"]

        def mock_retrbinary(command, callback):
            for chunk in mock_results:
                callback(chunk)

            return "226"

        mock_ftp = mock_ftp_class.return_value
        mock_ftp.retrbinary.side_effect = mock_retrbinary

        storage = storagelib.get_storage("ftp://user:[email protected]/some/dir/file")

        storage.save_to_filename(temp_output.name)

        self.assert_connected(mock_ftp_class, mock_ftp)

        mock_ftp.cwd.assert_called_with("some/dir")
        self.assertEqual(1, mock_ftp.retrbinary.call_count)
        self.assertEqual("RETR file", mock_ftp.retrbinary.call_args[0][0])

        with open(temp_output.name) as output_fp:
            self.assertEqual("foobar", output_fp.read())

    def __init__(self, uid, session):
        # load storage driver
        self.storage_type = 'awss3'
        self.storage = get_storage(self.storage_type)
        self.uid = uid

        self.session = session

    def test_connect_sets_tcp_keepalive_options_when_supported(self, mock_ftp_class, mock_socket):
        mock_socket.SOL_SOCKET = socket.SOL_SOCKET
        mock_socket.SOL_TCP = socket.SOL_TCP
        mock_socket.SO_KEEPALIVE = socket.SO_KEEPALIVE

        mock_socket.TCP_KEEPCNT = 1
        mock_socket.TCP_KEEPIDLE = 2
        mock_socket.TCP_KEEPINTVL = 3

        mock_ftp = mock_ftp_class.return_value
        in_file = StringIO("foobar")

        storage = storagelib.get_storage("ftp://user:[email protected]/some/dir/file")

        storage.load_from_file(in_file)

        mock_ftp.sock.setsockopt.assert_has_calls([
            mock.call(
                socket.SOL_SOCKET, socket.SO_KEEPALIVE,
                storagelib.storage.DEFAULT_FTP_KEEPALIVE_ENABLE),
            mock.call(
                socket.SOL_TCP, mock_socket.TCP_KEEPCNT, storagelib.storage.DEFAULT_FTP_KEEPCNT),
            mock.call(
                socket.SOL_TCP, mock_socket.TCP_KEEPIDLE, storagelib.storage.DEFAULT_FTP_KEEPIDLE),
            mock.call(
                socket.SOL_TCP, mock_socket.TCP_KEEPINTVL, storagelib.storage.DEFAULT_FTP_KEEPINTVL)
        ])

    def test_save_to_file(self, mock_session_class):
        mock_session = mock_session_class.return_value
        mock_s3 = mock_session.client.return_value

        mock_body = mock.Mock()
        mock_body.read.side_effect = [b"some", b"file", b"contents", None]
        mock_s3.get_object.return_value = {
            "Body": mock_body
        }

        mock_file = mock.Mock()

        storage = storagelib.get_storage(
            "s3://access_key:[email protected]/some/file?region=US_EAST")
        storage.save_to_file(mock_file)

        mock_session_class.assert_called_with(
            aws_access_key_id="access_key",
            aws_secret_access_key="access_secret",
            region_name="US_EAST")

        mock_session.client.assert_called_with("s3")
        mock_s3.get_object.assert_called_with(Bucket="bucket", Key="some/file")
        mock_file.write.assert_has_calls([
            mock.call(b"some"),
            mock.call(b"file"),
            mock.call(b"contents")
        ], any_order=False)

    def test_ftp_save_to_directory_creates_destination_directory_if_needed(
            self, mock_ftp_class, mock_path_exists, mock_makedirs, mock_chdir):
        mock_ftp = mock_ftp_class.return_value
        mock_ftp.pwd.return_value = "some/dir/file"

        # no files or folders
        mock_ftp.retrlines.side_effect = create_mock_ftp_directory_listing([])

        mock_path_exists.return_value = False

        storage = storagelib.get_storage("ftp://user:[email protected]/some/dir/file")

        storage.save_to_directory("/cat/pants")

        self.assert_connected(mock_ftp_class, mock_ftp)

        mock_ftp.cwd.assert_has_calls([
            mock.call("/some/dir/file"),
        ])

        mock_makedirs.assert_has_calls([
            mock.call("/cat/pants"),
        ])

        mock_chdir.assert_has_calls([
            mock.call("/cat/pants"),
        ])

        mock_ftp.retrbinary.assert_not_called()

    def test_save_to_file_with_specific_port(self, mock_ftp_class):
        out_file = StringIO()

        mock_results = ["foo", "bar"]

        def mock_retrbinary(command, callback):
            for chunk in mock_results:
                callback(chunk)

            return "226"

        mock_ftp = mock_ftp_class.return_value
        mock_ftp.retrbinary.side_effect = mock_retrbinary

        storage = storagelib.get_storage("ftp://user:[email protected]:12345/some/dir/file")

        storage.save_to_file(out_file)

        self.assert_connected(mock_ftp_class, mock_ftp, 12345)

        mock_ftp.cwd.assert_called_with("some/dir")
        self.assertEqual(1, mock_ftp.retrbinary.call_count)
        self.assertEqual("RETR file", mock_ftp.retrbinary.call_args[0][0])

        self.assertEqual("foobar", out_file.getvalue())

    def assert_transport_handles_directories(self, transport):
        variable = "TEST_STORAGE_{}_URI".format(transport)
        uri = os.getenv(variable, None)

        if not uri:
            raise unittest.SkipTest("Skipping {} - define {} to test".format(transport, variable))

        uri += "/{}".format(self.dest_prefix)
        parsed_url = urlparse(uri)
        print(
            "Testing using:", parsed_url.scheme, parsed_url.hostname, parsed_url.port,
            parsed_url.path)
        storage = get_storage(uri)

        print("Transport:", transport)
        print("\t* Uploading")
        storage.load_from_directory(self.directory)

        target_directory = tempfile.mkdtemp(prefix="dest-root")

        print("\t* Downloading")
        storage.save_to_directory(target_directory)

        print("\t* Checking")
        try:
            subprocess.check_output(
                ["diff", "-r", self.directory, target_directory], stderr=subprocess.STDOUT)
        except subprocess.CalledProcessError as error:
            print("Diff output:\n{}".format(error.output))
            raise

    def test_swift_load_from_directory(self, mock_timeout, mock_create_context):
        mock_swift = mock_create_context.return_value.get_client.return_value

        uri = "swift://{username}:{password}@{container}/{file}?" \
              "auth_endpoint={auth_endpoint}&region={region}" \
              "&tenant_id={tenant_id}".format(**self.params)

        storage = storagelib.get_storage(uri)

        temp_directory = create_temp_nested_directory_with_files()
        storage.load_from_directory(temp_directory["temp_directory"]["path"])

        self._assert_default_login_correct(mock_create_context, mock_timeout)

        mock_swift.upload_file.assert_has_calls([
            mock.call(
                self.params["container"], temp_directory["temp_input_two"]["path"],
                os.path.join(self.params["file"], temp_directory["temp_input_two"]["name"])),
            mock.call(
                self.params["container"], temp_directory["temp_input_one"]["path"],
                os.path.join(self.params["file"], temp_directory["temp_input_one"]["name"])),
            mock.call(
                self.params["container"], temp_directory["nested_temp_input"]["path"],
                os.path.join(
                    self.params["file"], temp_directory["nested_temp_directory"]["name"],
                    temp_directory["nested_temp_input"]["name"]))
        ], any_order=True)

    def test_ftp_delete_directory(self, mock_ftp_class):
        mock_ftp = mock_ftp_class.return_value

        mock_ftp.pwd.return_value = "some/dir/file"

        mock_ftp.retrlines.side_effect = create_mock_ftp_directory_listing([
            # root
            [
                "drwxrwxr-x 3 test test 4.0K Apr  9 10:54 dir1",
                "drwxrwxr-x 3 test test 4.0K Apr  9 10:54 dir2",
                "-rwxrwxr-x 3 test test 4.0K Apr  9 10:54 file1",
                "-rwxrwxr-x 3 test test 4.0K Apr  9 10:54 file2",
            ],
            # dir1
            [
                "drwxrwxr-x 3 test test 4.0K Apr  9 10:54 dir with spaces",
                "-rwxrwxr-x 3 test test 4.0K Apr  9 10:54 file3",
            ],
            # dir with spaces
            [
                "-rwxrwxr-x 3 test test 4.0K Apr  9 10:54 file with spaces"
            ],
            # dir2
            [
                "drwxrwxr-x 3 test test 4.0K Apr  9 10:54 dir4",
            ],
        ])

        storage = storagelib.get_storage("ftp://user:[email protected]/some/dir/file")
        storage.delete_directory()

        self.assert_connected(mock_ftp_class, mock_ftp)

        mock_ftp.cwd.assert_has_calls([
            mock.call("/some/dir/file"),
            mock.call("some/dir/file/dir1"),
            mock.call("some/dir/file/dir1/dir with spaces"),
            mock.call("some/dir/file/dir2"),
            mock.call("some/dir/file/dir2/dir4")
        ])
        self.assertEqual(5, mock_ftp.cwd.call_count)

        mock_ftp.delete.assert_has_calls([
            mock.call("/some/dir/file/dir1/dir with spaces/file with spaces"),
            mock.call("/some/dir/file/dir1/file3"),
            mock.call("/some/dir/file/file2"),
            mock.call("/some/dir/file/file1")
        ], any_order=True)
        self.assertEqual(4, mock_ftp.delete.call_count)

        mock_ftp.rmd.assert_has_calls([
            mock.call("/some/dir/file/dir2/dir4"),
            mock.call("/some/dir/file/dir2"),
            mock.call("/some/dir/file/dir1/dir with spaces"),
            mock.call("/some/dir/file/dir1"),
            mock.call("/some/dir/file")
        ])
        self.assertEqual(5, mock_ftp.rmd.call_count)

    def test_ftp_get_download_url_returns_none_with_empty_base(self, mock_ftp_class):
        ftpuri = "ftp://user:[email protected]/some/dir/file.txt"

        storage = storagelib.get_storage(ftpuri)

        with self.assertRaises(DownloadUrlBaseUndefinedError):
            storage.get_download_url()

        mock_ftp_class.assert_not_called()

    def test_load_from_file_creates_intermediate_dirs(self, mock_exists, mock_makedirs, mock_copy):
        mock_exists.return_value = False

        storage = storagelib.get_storage("file:///foo/bar/file")
        storage.load_from_filename("input_file")

        mock_exists.assert_called_with("/foo/bar")
        mock_makedirs.assert_called_with("/foo/bar")
        mock_copy.assert_called_with("input_file", "/foo/bar/file")

    def test_local_storage_delete_directory(self, mock_remove, mock_rmtree):
        temp_directory = create_temp_nested_directory_with_files()

        storage = storagelib.get_storage(
            "file://{0}".format(temp_directory["temp_directory"]["path"]))
        storage.delete_directory()

        self.assertFalse(mock_remove.called)
        mock_rmtree.assert_called_once_with(temp_directory["temp_directory"]["path"], True)

    def test_local_storage_load_from_file(self):
        in_file = StringIO("foobar")
        temp_output = tempfile.NamedTemporaryFile()

        storage = storagelib.get_storage("file://{0}".format(temp_output.name))
        storage.load_from_file(in_file)

        with open(temp_output.name) as temp_output_fp:
            self.assertEqual("foobar", temp_output_fp.read())

    def __init__(self, uid):
        # load storage driver
        self.storage = get_storage('mock', uid=uid)
        self.uid = uid

        engine = create_engine('sqlite:///C:\\Users\\xiang\\test.db')

        Session = sessionmaker(bind=engine)
        self.session = Session()