test_utils.py

importwarnings
importpytest
fromscipy.linalgimporteigh, pinvh
fromcollectionsimportnamedtuple
importnumpyasnp
fromnumpy.testingimportassert_array_equal, assert_equal
fromsklearn.model_selectionimporttrain_test_split
fromsklearn.utilsimportcheck_random_state, shuffle
frommetric_learn.sklearn_shimsimportset_random_state
fromsklearn.baseimportclone
frommetric_learn._utilimport (check_input, make_context, preprocess_tuples,
make_name, preprocess_points,
check_collapsed_pairs, validate_vector,
_check_sdp_from_eigen, _check_n_components,
check_y_valid_values_for_pairs,
_auto_select_init, _pseudo_inverse_from_eig)
frommetric_learnimport (ITML, LSML, MMC, RCA, SDML, Covariance, LFDA,
LMNN, MLKR, NCA, ITML_Supervised, LSML_Supervised,
MMC_Supervised, RCA_Supervised, SDML_Supervised,
SCML, SCML_Supervised, Constraints)
frommetric_learn.base_metricimport (ArrayIndexer, MahalanobisMixin,
_PairsClassifierMixin,
_TripletsClassifierMixin,
_QuadrupletsClassifierMixin)
frommetric_learn.exceptionsimportPreprocessorError, NonPSDError
fromsklearn.datasetsimportmake_regression, make_blobs, load_iris


SEED=42
RNG=check_random_state(SEED)

Dataset=namedtuple('Dataset', ('data target preprocessor to_transform'))
# Data and target are what we will fit on. Preprocessor is the additional
# data if we use a preprocessor (which should be the default ArrayIndexer),
# and to_transform is some additional data that we would want to transform


defbuild_classification(with_preprocessor=False):
"""Basic array for testing when using a preprocessor"""
X, y=shuffle(*make_blobs(random_state=SEED),
random_state=SEED)
indices=shuffle(np.arange(X.shape[0]), random_state=SEED).astype(int)
ifwith_preprocessor:
returnDataset(indices, y[indices], X, indices)
else:
returnDataset(X[indices], y[indices], None, X[indices])


defbuild_regression(with_preprocessor=False):
"""Basic array for testing when using a preprocessor"""
X, y=shuffle(*make_regression(n_samples=100, n_features=5,
random_state=SEED),
random_state=SEED)
indices=shuffle(np.arange(X.shape[0]), random_state=SEED).astype(int)
ifwith_preprocessor:
returnDataset(indices, y[indices], X, indices)
else:
returnDataset(X[indices], y[indices], None, X[indices])


defbuild_data():
input_data, labels=load_iris(return_X_y=True)
X, y=shuffle(input_data, labels, random_state=SEED)
n_constraints=50
constraints=Constraints(y)
pairs= (
constraints
 .positive_negative_pairs(n_constraints, same_length=True,
random_state=check_random_state(SEED)))
returnX, pairs


defbuild_pairs(with_preprocessor=False):
# builds a toy pairs problem
X, indices=build_data()
c=np.vstack([np.column_stack(indices[:2]), np.column_stack(indices[2:])])
target=np.concatenate([np.ones(indices[0].shape[0]),
-np.ones(indices[0].shape[0])])
c, target=shuffle(c, target, random_state=SEED)
ifwith_preprocessor:
# if preprocessor, we build a 2D array of pairs of indices
returnDataset(c, target, X, c[:, 0])
else:
# if not, we build a 3D array of pairs of samples
returnDataset(X[c], target, None, X[c[:, 0]])


defbuild_triplets(with_preprocessor=False):
input_data, labels=load_iris(return_X_y=True)
X, y=shuffle(input_data, labels, random_state=SEED)
constraints=Constraints(y)
triplets=constraints.generate_knntriplets(X, k_genuine=3, k_impostor=4)
ifwith_preprocessor:
# if preprocessor, we build a 2D array of triplets of indices
returnDataset(triplets, np.ones(len(triplets)), X, np.arange(len(X)))
else:
# if not, we build a 3D array of triplets of samples
returnDataset(X[triplets], np.ones(len(triplets)), None, X)


defbuild_quadruplets(with_preprocessor=False):
# builds a toy quadruplets problem
X, indices=build_data()
c=np.column_stack(indices)
target=np.ones(c.shape[0]) # quadruplets targets are not used
# anyways
c, target=shuffle(c, target, random_state=SEED)
ifwith_preprocessor:
# if preprocessor, we build a 2D array of quadruplets of indices
returnDataset(c, target, X, c[:, 0])
else:
# if not, we build a 3D array of quadruplets of samples
returnDataset(X[c], target, None, X[c[:, 0]])


quadruplets_learners= [(LSML(), build_quadruplets)]
ids_quadruplets_learners=list(map(lambdax: x.__class__.__name__,
 [learnerfor (learner, _) in
quadruplets_learners]))

triplets_learners= [(SCML(n_basis=320), build_triplets)]
ids_triplets_learners=list(map(lambdax: x.__class__.__name__,
 [learnerfor (learner, _) in
triplets_learners]))

pairs_learners= [(ITML(max_iter=2), build_pairs), # max_iter=2 to be faster
 (MMC(max_iter=2), build_pairs), # max_iter=2 to be faster
 (SDML(prior='identity', balance_param=1e-5), build_pairs)]
ids_pairs_learners=list(map(lambdax: x.__class__.__name__,
 [learnerfor (learner, _) in
pairs_learners]))

classifiers= [(Covariance(), build_classification),
 (LFDA(), build_classification),
 (LMNN(), build_classification),
 (NCA(), build_classification),
 (RCA(), build_classification),
 (ITML_Supervised(max_iter=5), build_classification),
 (LSML_Supervised(), build_classification),
 (MMC_Supervised(max_iter=5), build_classification),
 (RCA_Supervised(n_chunks=5), build_classification),
 (SDML_Supervised(prior='identity', balance_param=1e-5),
build_classification),
 (SCML_Supervised(n_basis=80), build_classification)]
ids_classifiers=list(map(lambdax: x.__class__.__name__,
 [learnerfor (learner, _) in
classifiers]))

regressors= [(MLKR(init='pca'), build_regression)]
ids_regressors=list(map(lambdax: x.__class__.__name__,
 [learnerfor (learner, _) inregressors]))

WeaklySupervisedClasses= (_PairsClassifierMixin,
_TripletsClassifierMixin,
_QuadrupletsClassifierMixin)

tuples_learners=pairs_learners+triplets_learners+quadruplets_learners
ids_tuples_learners=ids_pairs_learners+ids_triplets_learners \
+ids_quadruplets_learners

supervised_learners=classifiers+regressors
ids_supervised_learners=ids_classifiers+ids_regressors

metric_learners=tuples_learners+supervised_learners
ids_metric_learners=ids_tuples_learners+ids_supervised_learners

metric_learners_pipeline=pairs_learners+supervised_learners
ids_metric_learners_pipeline=ids_pairs_learners+ids_supervised_learners


defremove_y(estimator, X, y):
"""Quadruplets and triplets learners have no y in fit, but to write test for
 all estimators, it is convenient to have this function, that will return X
 and y if the estimator needs a y to fit on, and just X otherwise."""
no_y_fit=quadruplets_learners+triplets_learners
ifestimator.__class__.__name__in [e.__class__.__name__
for (e, _) inno_y_fit]:
return (X,)
else:
return (X, y)


defmock_preprocessor(indices):
"""A preprocessor for testing purposes that returns an all ones 3D array
 """
returnnp.ones((indices.shape[0], 3))


@pytest.mark.parametrize('type_of_inputs', ['other', 'tuple', 'classics', 2,
int, NCA()])
deftest_check_input_invalid_type_of_inputs(type_of_inputs):
"""Tests that an invalid type of inputs in check_inputs raises an error."""
withpytest.raises(ValueError) ase:
check_input([[0.2, 2.1], [0.2, .8]], type_of_inputs=type_of_inputs)
msg= ("Unknown value {} for type_of_inputs. Valid values are "
"'classic' or 'tuples'.".format(type_of_inputs))
assertstr(e.value) ==msg


# ---------------- test check_input with 'tuples' type_of_input' ------------


deftuples_prep():
"""Basic array for testing when using a preprocessor"""
tuples=np.array([[1, 2],
 [2, 3]])
returntuples


deftuples_no_prep():
"""Basic array for testing when using no preprocessor"""
tuples=np.array([[[1., 2.3], [2.3, 5.3]],
 [[2.3, 4.3], [0.2, 0.4]]])
returntuples


@pytest.mark.parametrize('estimator, expected',
 [(NCA(), " by NCA"), ('NCA', " by NCA"), (None, "")])
deftest_make_context(estimator, expected):
"""test the make_name function"""
assertmake_context(estimator) ==expected


@pytest.mark.parametrize('estimator, expected',
 [(NCA(), "NCA"), ('NCA', "NCA"), (None, None)])
deftest_make_name(estimator, expected):
"""test the make_name function"""
assertmake_name(estimator) ==expected


@pytest.mark.parametrize('estimator, context',
 [(NCA(), " by NCA"), ('NCA', " by NCA"), (None, "")])
@pytest.mark.parametrize('load_tuples, preprocessor',
 [(tuples_prep, mock_preprocessor),
 (tuples_no_prep, None),
 (tuples_no_prep, mock_preprocessor)])
deftest_check_tuples_invalid_tuple_size(estimator, context, load_tuples,
preprocessor):
"""Checks that the exception are raised if tuple_size is not the one
 expected"""
tuples=load_tuples()
preprocessed_tuples= (preprocess_tuples(tuples, preprocessor)
if (preprocessorisnotNoneand
tuples.ndim==2) elsetuples)
expected_msg= ("Tuples of 3 element(s) expected{}. Got tuples of 2 "
"element(s) instead (shape={}):\ninput={}.\n"
 .format(context, preprocessed_tuples.shape,
preprocessed_tuples))
withpytest.raises(ValueError) asraised_error:
check_input(tuples, type_of_inputs='tuples', tuple_size=3,
preprocessor=preprocessor, estimator=estimator)
assertstr(raised_error.value) ==expected_msg


@pytest.mark.parametrize('estimator, context',
 [(NCA(), " by NCA"), ('NCA', " by NCA"), (None, "")])
@pytest.mark.parametrize('tuples, found, expected, preprocessor',
 [(5, '0', '2D array of indicators or 3D array of '
'formed tuples', mock_preprocessor),
 (5, '0', '3D array of formed tuples', None),
 ([1, 2], '1', '2D array of indicators or 3D array '
'of formed tuples', mock_preprocessor),
 ([1, 2], '1', '3D array of formed tuples', None),
 ([[[[5]]]], '4', '2D array of indicators or 3D array'
' of formed tuples',
mock_preprocessor),
 ([[[[5]]]], '4', '3D array of formed tuples', None),
 ([[1], [3]], '2', '3D array of formed '
'tuples', None)])
deftest_check_tuples_invalid_shape(estimator, context, tuples, found,
expected, preprocessor):
"""Checks that a value error with the appropriate message is raised if
 shape is invalid (not 2D with preprocessor or 3D with no preprocessor)
 """
tuples=np.array(tuples)
msg= ("{} expected{}{}. Found {}D array instead:\ninput={}. Reshape your "
"data{}.\n"
 .format(expected, context, ' when using a preprocessor'
ifpreprocessorelse'', found, tuples,
' and/or use a preprocessor'if
 (notpreprocessorandtuples.ndim==2) else''))
withpytest.raises(ValueError) asraised_error:
check_input(tuples, type_of_inputs='tuples',
preprocessor=preprocessor, ensure_min_samples=0,
estimator=estimator)
assertstr(raised_error.value) ==msg


@pytest.mark.parametrize('estimator, context',
 [(NCA(), " by NCA"), ('NCA', " by NCA"), (None, "")])
deftest_check_tuples_invalid_n_features(estimator, context):
"""Checks that the right warning is printed if not enough features
 Here we only test if no preprocessor (otherwise we don't ensure this)
 """
msg= ("Found array with 2 feature(s) (shape={}) while"
" a minimum of 3 is required{}.".format(tuples_no_prep().shape,
context))
withpytest.raises(ValueError) asraised_error:
check_input(tuples_no_prep(), type_of_inputs='tuples',
preprocessor=None, ensure_min_features=3,
estimator=estimator)
assertstr(raised_error.value) ==msg


@pytest.mark.parametrize('estimator, context',
 [(NCA(), " by NCA"), ('NCA', " by NCA"), (None, "")])
@pytest.mark.parametrize('load_tuples, preprocessor',
 [(tuples_prep, mock_preprocessor),
 (tuples_no_prep, None),
 (tuples_no_prep, mock_preprocessor)])
deftest_check_tuples_invalid_n_samples(estimator, context, load_tuples,
preprocessor):
"""Checks that the right warning is printed if n_samples is too small"""
tuples=load_tuples()
msg= ("Found array with 2 sample(s) (shape={}) while a minimum of 3 "
"is required{}.".format((preprocess_tuples(tuples, preprocessor)
if (preprocessorisnotNoneand
tuples.ndim==2) elsetuples).shape,
context))
withpytest.raises(ValueError) asraised_error:
check_input(tuples, type_of_inputs='tuples',
preprocessor=preprocessor,
ensure_min_samples=3, estimator=estimator)
assertstr(raised_error.value) ==msg


deftest_check_tuples_invalid_dtype_not_convertible_with_preprocessor():
"""Checks that a value error is thrown if attempting to convert an
 input not convertible to float, when using a preprocessor
 """

defpreprocessor(indices):
# preprocessor that returns objects
returnnp.full((indices.shape[0], 3), 'a')

withpytest.raises(ValueError):
check_input(tuples_prep(), type_of_inputs='tuples',
preprocessor=preprocessor, dtype=np.float64)


deftest_check_tuples_invalid_dtype_not_convertible_without_preprocessor():
"""Checks that a value error is thrown if attempting to convert an
 input not convertible to float, when using no preprocessor
 """
tuples=np.full_like(tuples_no_prep(), 'a', dtype=object)
withpytest.raises(ValueError):
check_input(tuples, type_of_inputs='tuples',
preprocessor=None, dtype=np.float64)


@pytest.mark.parametrize('tuple_size', [2, None])
deftest_check_tuples_valid_tuple_size(tuple_size):
"""For inputs that have the right matrix dimension (2D or 3D for instance),
 checks that checking the number of tuples (pairs, quadruplets, etc) raises
 no warning if there is the right number of points in a tuple.
 """
withwarnings.catch_warnings(record=True) asrecord:
check_input(tuples_prep(), type_of_inputs='tuples',
preprocessor=mock_preprocessor, tuple_size=tuple_size)
check_input(tuples_no_prep(), type_of_inputs='tuples', preprocessor=None,
tuple_size=tuple_size)
assertlen(record) ==0


@pytest.mark.parametrize('tuples',
 [np.array([[2.5, 0.1, 2.6],
 [1.6, 4.8, 9.1]]),
np.array([[2, 0, 2],
 [1, 4, 9]]),
np.array([["img1.png", "img3.png"],
 ["img2.png", "img4.png"]]),
 [[2, 0, 2],
 [1, 4, 9]],
 [np.array([2, 0, 2]),
np.array([1, 4, 9])],
 ((2, 0, 2),
 (1, 4, 9)),
np.array([[[1.2, 2.2], [1.4, 3.3]],
 [[2.6, 2.3], [3.4, 5.0]]])])
deftest_check_tuples_valid_with_preprocessor(tuples):
"""Test that valid inputs when using a preprocessor raises no warning"""
withwarnings.catch_warnings(record=True) asrecord:
check_input(tuples, type_of_inputs='tuples',
preprocessor=mock_preprocessor)
assertlen(record) ==0


@pytest.mark.parametrize('tuples',
 [np.array([[[2.5], [0.1], [2.6]],
 [[1.6], [4.8], [9.1]],
 [[5.6], [2.8], [6.1]]]),
np.array([[[2], [0], [2]],
 [[1], [4], [9]],
 [[1], [5], [3]]]),
 [[[2], [0], [2]],
 [[1], [4], [9]],
 [[3], [4], [29]]],
 (((2, 1), (0, 2), (2, 3)),
 ((1, 2), (4, 4), (9, 3)),
 ((3, 1), (4, 4), (29, 4)))])
deftest_check_tuples_valid_without_preprocessor(tuples):
"""Test that valid inputs when using no preprocessor raises no warning"""
withwarnings.catch_warnings(record=True) asrecord:
check_input(tuples, type_of_inputs='tuples', preprocessor=None)
assertlen(record) ==0


deftest_check_tuples_behaviour_auto_dtype():
"""Checks that check_tuples allows by default every type if using a
 preprocessor, and numeric types if using no preprocessor"""
tuples_prep= [['img1.png', 'img2.png'], ['img3.png', 'img5.png']]
withwarnings.catch_warnings(record=True) asrecord:
check_input(tuples_prep, type_of_inputs='tuples',
preprocessor=mock_preprocessor)
assertlen(record) ==0

withwarnings.catch_warnings(record=True) asrecord:
check_input(tuples_no_prep(), type_of_inputs='tuples') # numeric type
assertlen(record) ==0

# not numeric type
tuples_no_prep_bis=np.array([[['img1.png'], ['img2.png']],
 [['img3.png'], ['img5.png']]])
tuples_no_prep_bis=tuples_no_prep_bis.astype(object)
withpytest.raises(ValueError):
check_input(tuples_no_prep_bis, type_of_inputs='tuples')


deftest_check_tuples_invalid_complex_data():
"""Checks that the right error message is thrown if given complex data (
 this comes from sklearn's check_array's message)"""
tuples=np.array([[[1+2j, 3+4j], [5+7j, 5+7j]],
 [[1+3j, 2+4j], [5+8j, 1+7j]]])
msg= ("Complex data not supported\n"
"{}\n".format(tuples))
withpytest.raises(ValueError) asraised_error:
check_input(tuples, type_of_inputs='tuples')
assertstr(raised_error.value) ==msg


# ------------- test check_input with 'classic' type_of_inputs ----------------


defpoints_prep():
"""Basic array for testing when using a preprocessor"""
points=np.array([1, 2])
returnpoints


defpoints_no_prep():
"""Basic array for testing when using no preprocessor"""
points=np.array([[1., 2.3],
 [2.3, 4.3]])
returnpoints


@pytest.mark.parametrize('estimator, context',
 [(NCA(), " by NCA"), ('NCA', " by NCA"), (None, "")])
@pytest.mark.parametrize('points, found, expected, preprocessor',
 [(5, '0', '1D array of indicators or 2D array of '
'formed points', mock_preprocessor),
 (5, '0', '2D array of formed points', None),
 ([1, 2], '1', '2D array of formed points', None),
 ([[[5]]], '3', '1D array of indicators or 2D '
'array of formed points',
mock_preprocessor),
 ([[[5]]], '3', '2D array of formed points', None)])
deftest_check_classic_invalid_shape(estimator, context, points, found,
expected, preprocessor):
"""Checks that a value error with the appropriate message is raised if
 shape is invalid (valid being 1D or 2D with preprocessor or 2D with no
 preprocessor)
 """
points=np.array(points)
msg= ("{} expected{}{}. Found {}D array instead:\ninput={}. Reshape your "
"data{}.\n"
 .format(expected, context, ' when using a preprocessor'
ifpreprocessorelse'', found, points,
' and/or use a preprocessor'if
 (notpreprocessorandpoints.ndim==1) else''))
withpytest.raises(ValueError) asraised_error:
check_input(points, type_of_inputs='classic', preprocessor=preprocessor,
ensure_min_samples=0,
estimator=estimator)
assertstr(raised_error.value) ==msg


@pytest.mark.parametrize('estimator, context',
 [(NCA(), " by NCA"), ('NCA', " by NCA"), (None, "")])
deftest_check_classic_invalid_n_features(estimator, context):
"""Checks that the right warning is printed if not enough features
 Here we only test if no preprocessor (otherwise we don't ensure this)
 """
msg= ("Found array with 2 feature(s) (shape={}) while"
" a minimum of 3 is required{}.".format(points_no_prep().shape,
context))
withpytest.raises(ValueError) asraised_error:
check_input(points_no_prep(), type_of_inputs='classic',
preprocessor=None, ensure_min_features=3,
estimator=estimator)
assertstr(raised_error.value) ==msg


@pytest.mark.parametrize('estimator, context',
 [(NCA(), " by NCA"), ('NCA', " by NCA"), (None, "")])
@pytest.mark.parametrize('load_points, preprocessor',
 [(points_prep, mock_preprocessor),
 (points_no_prep, None),
 (points_no_prep, mock_preprocessor)])
deftest_check_classic_invalid_n_samples(estimator, context, load_points,
preprocessor):
"""Checks that the right warning is printed if n_samples is too small"""
points=load_points()
msg= ("Found array with 2 sample(s) (shape={}) while a minimum of 3 "
"is required{}.".format((preprocess_points(points,
preprocessor)
ifpreprocessorisnotNoneand
points.ndim==1else
points).shape,
context))
withpytest.raises(ValueError) asraised_error:
check_input(points, type_of_inputs='classic', preprocessor=preprocessor,
ensure_min_samples=3,
estimator=estimator)
assertstr(raised_error.value) ==msg


@pytest.mark.parametrize('preprocessor, points',
 [(mock_preprocessor, np.array([['a', 'b'],
 ['e', 'b']])),
 (None, np.array([[['b', 'v'], ['a', 'd']],
 [['x', 'u'], ['c', 'a']]]))])
deftest_check_classic_invalid_dtype_not_convertible(preprocessor, points):
"""Checks that a value error is thrown if attempting to convert an
 input not convertible to float
 """
withpytest.raises(ValueError):
check_input(points, type_of_inputs='classic',
preprocessor=preprocessor, dtype=np.float64)


@pytest.mark.parametrize('points',
 [["img1.png", "img3.png", "img2.png"],
np.array(["img1.png", "img3.png", "img2.png"]),
 [2, 0, 2, 1, 4, 9],
range(10),
np.array([2, 0, 2]),
 (2, 0, 2),
np.array([[1.2, 2.2],
 [2.6, 2.3]])])
deftest_check_classic_valid_with_preprocessor(points):
"""Test that valid inputs when using a preprocessor raises no warning"""
withwarnings.catch_warnings(record=True) asrecord:
check_input(points, type_of_inputs='classic',
preprocessor=mock_preprocessor)
assertlen(record) ==0


@pytest.mark.parametrize('points',
 [np.array([[2.5, 0.1, 2.6],
 [1.6, 4.8, 9.1],
 [5.6, 2.8, 6.1]]),
np.array([[2, 0, 2],
 [1, 4, 9],
 [1, 5, 3]]),
 [[2, 0, 2],
 [1, 4, 9],
 [3, 4, 29]],
 ((2, 1, 0, 2, 2, 3),
 (1, 2, 4, 4, 9, 3),
 (3, 1, 4, 4, 29, 4))])
deftest_check_classic_valid_without_preprocessor(points):
"""Test that valid inputs when using no preprocessor raises no warning"""
withwarnings.catch_warnings(record=True) asrecord:
check_input(points, type_of_inputs='classic', preprocessor=None)
assertlen(record) ==0


deftest_check_classic_by_default():
"""Checks that 'classic' is the default behaviour of check_input"""
assert (check_input([[2, 3], [3, 2]]) ==
check_input([[2, 3], [3, 2]], type_of_inputs='classic')).all()


deftest_check_classic_behaviour_auto_dtype():
"""Checks that check_input (for points) allows by default every type if
 using a preprocessor, and numeric types if using no preprocessor"""
points_prep= ['img1.png', 'img2.png', 'img3.png', 'img5.png']
withwarnings.catch_warnings(record=True) asrecord:
check_input(points_prep, type_of_inputs='classic',
preprocessor=mock_preprocessor)
assertlen(record) ==0

withwarnings.catch_warnings(record=True) asrecord:
check_input(points_no_prep(), type_of_inputs='classic') # numeric type
assertlen(record) ==0

# not numeric type
points_no_prep_bis=np.array(['img1.png', 'img2.png', 'img3.png',
'img5.png'])
points_no_prep_bis=points_no_prep_bis.astype(object)
withpytest.raises(ValueError):
check_input(points_no_prep_bis, type_of_inputs='classic')


deftest_check_classic_invalid_complex_data():
"""Checks that the right error message is thrown if given complex data (
 this comes from sklearn's check_array's message)"""
points=np.array([[[1+2j, 3+4j], [5+7j, 5+7j]],
 [[1+3j, 2+4j], [5+8j, 1+7j]]])
msg= ("Complex data not supported\n"
"{}\n".format(points))
withpytest.raises(ValueError) asraised_error:
check_input(points, type_of_inputs='classic')
assertstr(raised_error.value) ==msg


# ----------------------------- Test preprocessor -----------------------------


X=np.array([[0.89, 0.11, 1.48, 0.12],
 [2.63, 1.08, 1.68, 0.46],
 [1.00, 0.59, 0.62, 1.15]])


classMockFileLoader:
"""Preprocessor that takes a root file path at construction and simulates
 fetching the file in the specific root folder when given the name of the
 file"""

def__init__(self, root):
self.root=root
self.folders= {'fake_root': {'img0.png': X[0],
'img1.png': X[1],
'img2.png': X[2]
 },
'other_folder': {} # empty folder
 }

def__call__(self, path_list):
images=list()
forpathinpath_list:
images.append(self.folders[self.root][path])
returnnp.array(images)


defmock_id_loader(list_of_indicators):
"""A preprocessor as a function that takes indicators (strings) and
 returns the corresponding samples"""
points= []
forindicatorinlist_of_indicators:
points.append(X[int(indicator[2:])])
returnnp.array(points)


tuples_list= [np.array([[0, 1],
 [2, 1]]),

np.array([['img0.png', 'img1.png'],
 ['img2.png', 'img1.png']]),

np.array([['id0', 'id1'],
 ['id2', 'id1']])
 ]

points_list= [np.array([0, 1, 2, 1]),

np.array(['img0.png', 'img1.png', 'img2.png', 'img1.png']),

np.array(['id0', 'id1', 'id2', 'id1'])
 ]

preprocessors= [X, MockFileLoader('fake_root'), mock_id_loader]


@pytest.fixture
defy_tuples():
y= [-1, 1]
returny


@pytest.fixture
defy_points():
y= [0, 1, 0, 0]
returny


@pytest.mark.parametrize('preprocessor, tuples', zip(preprocessors,
tuples_list))
deftest_preprocessor_weakly_supervised(preprocessor, tuples, y_tuples):
"""Tests different ways to use the preprocessor argument: an array,
 a class callable, and a function callable, with a weakly supervised
 algorithm
 """
nca=ITML(preprocessor=preprocessor)
nca.fit(tuples, y_tuples)


@pytest.mark.parametrize('preprocessor, points', zip(preprocessors,
points_list))
deftest_preprocessor_supervised(preprocessor, points, y_points):
"""Tests different ways to use the preprocessor argument: an array,
 a class callable, and a function callable, with a supervised algorithm
 """
lfda=LFDA(preprocessor=preprocessor)
lfda.fit(points, y_points)


@pytest.mark.parametrize('estimator', ['NCA', NCA(), None])
deftest_preprocess_tuples_invalid_message(estimator):
"""Checks that if the preprocessor does some weird stuff, the preprocessed
 input is detected as weird. Checks this for preprocess_tuples."""

context=make_context(estimator) + (' after the preprocessor '
'has been applied')

defpreprocessor(sequence):
returnnp.ones((len(sequence), 2, 2)) # returns a 3D array instead of 2D

withpytest.raises(ValueError) asraised_error:
check_input(np.ones((3, 2)), type_of_inputs='tuples',
preprocessor=preprocessor, estimator=estimator)
expected_msg= ("3D array of formed tuples expected{}. Found 4D "
"array instead:\ninput={}. Reshape your data{}.\n"
 .format(context, np.ones((3, 2, 2, 2)),
' and/or use a preprocessor'ifpreprocessor
isnotNoneelse''))
assertstr(raised_error.value) ==expected_msg


@pytest.mark.parametrize('estimator', ['NCA', NCA(), None])
deftest_preprocess_points_invalid_message(estimator):
"""Checks that if the preprocessor does some weird stuff, the preprocessed
 input is detected as weird."""

context=make_context(estimator) + (' after the preprocessor '
'has been applied')

defpreprocessor(sequence):
returnnp.ones((len(sequence), 2, 2)) # returns a 3D array instead of 2D

withpytest.raises(ValueError) asraised_error:
check_input(np.ones((3,)), type_of_inputs='classic',
preprocessor=preprocessor, estimator=estimator)
expected_msg= ("2D array of formed points expected{}. "
"Found 3D array instead:\ninput={}. Reshape your data{}.\n"
 .format(context, np.ones((3, 2, 2)),
' and/or use a preprocessor'ifpreprocessor
isnotNoneelse''))
assertstr(raised_error.value) ==expected_msg


deftest_preprocessor_error_message():
"""Tests whether the preprocessor returns a preprocessor error when there
 is a problem using the preprocessor
 """
preprocessor=ArrayIndexer(np.array([[1.2, 3.3], [3.1, 3.2]]))

# with tuples
X=np.array([[[2, 3], [3, 3]], [[2, 3], [3, 2]]])
# There are less samples than the max index we want to preprocess
withpytest.raises(PreprocessorError):
preprocess_tuples(X, preprocessor)

# with points
X=np.array([[1], [2], [3], [3]])
withpytest.raises(PreprocessorError):
preprocess_points(X, preprocessor)


@pytest.mark.parametrize('input_data', [[[5, 3], [3, 2]],
 ((5, 3), (3, 2))
 ])
@pytest.mark.parametrize('indices', [[0, 1], (1, 0)])
deftest_array_like_indexer_array_like_valid_classic(input_data, indices):
"""Checks that any array-like is valid in the 'preprocessor' argument,
 and in the indices, for a classic input"""
classMockMetricLearner(MahalanobisMixin):
deffit(self):
pass
pass

mock_algo=MockMetricLearner(preprocessor=input_data)
mock_algo._prepare_inputs(indices, type_of_inputs='classic')


@pytest.mark.parametrize('input_data', [[[5, 3], [3, 2]],
 ((5, 3), (3, 2))
 ])
@pytest.mark.parametrize('indices', [[[0, 1], [1, 0]], ((1, 0), (1, 0))])
deftest_array_like_indexer_array_like_valid_tuples(input_data, indices):
"""Checks that any array-like is valid in the 'preprocessor' argument,
 and in the indices, for a classic input"""
classMockMetricLearner(MahalanobisMixin):
deffit(self):
pass
pass

mock_algo=MockMetricLearner(preprocessor=input_data)
mock_algo._prepare_inputs(indices, type_of_inputs='tuples')


@pytest.mark.parametrize('preprocessor', [4, NCA()])
deftest_error_message_check_preprocessor(preprocessor):
"""Checks that if the preprocessor given is not an array-like or a
 callable, the right error message is returned"""
classMockMetricLearner(MahalanobisMixin):
pass

mock_algo=MockMetricLearner(preprocessor=preprocessor)
withpytest.raises(ValueError) ase:
mock_algo._check_preprocessor()
assertstr(e.value) == ("Invalid type for the preprocessor: {}. You should "
"provide either None, an array-like object, "
"or a callable.".format(type(preprocessor)))


@pytest.mark.parametrize('estimator, _', tuples_learners,
ids=ids_tuples_learners)
deftest_error_message_tuple_size(estimator, _):
"""Tests that if a tuples learner is not given the good number of points
 per tuple, it throws an error message"""
estimator=clone(estimator)
set_random_state(estimator)
invalid_pairs=np.ones((2, 5, 2))
y= [1, 1]
withpytest.raises(ValueError) asraised_err:
estimator.fit(*remove_y(estimator, invalid_pairs, y))
expected_msg= ("Tuples of {} element(s) expected{}. Got tuples of 5 "
"element(s) instead (shape=(2, 5, 2)):\ninput={}.\n"
 .format(estimator._tuple_size, make_context(estimator),
invalid_pairs))
assertstr(raised_err.value) ==expected_msg


@pytest.mark.parametrize('estimator, _', metric_learners,
ids=ids_metric_learners)
deftest_error_message_t_pair_distance_or_score(estimator, _):
"""Tests that if you want to pair_distance or pair_score on triplets
 for instance, it returns the right error message
 """
estimator=clone(estimator)
set_random_state(estimator)
estimator._check_preprocessor()
triplets=np.array([[[1.3, 6.3], [3., 6.8], [6.5, 4.4]],
 [[1.9, 5.3], [1., 7.8], [3.2, 1.2]]])
withpytest.raises(ValueError) asraised_err:
estimator.pair_score(triplets)
expected_msg= ("Tuples of 2 element(s) expected{}. Got tuples of 3 "
"element(s) instead (shape=(2, 3, 2)):\ninput={}.\n"
 .format(make_context(estimator), triplets))
assertstr(raised_err.value) ==expected_msg

not_implemented_msg=""
# Todo in 0.7.0: Change 'not_implemented_msg' for the message that says
# "This learner does not have pair_distance"

# One exception will trigger for sure
withpytest.raises(Exception) asraised_exception:
estimator.pair_distance(triplets)
err_value=raised_exception.value.args[0]
asserterr_value==expected_msgorerr_value==not_implemented_msg


deftest_preprocess_tuples_simple_example():
"""Test the preprocessor on a very simple example of tuples to ensure the
 result is as expected"""
array=np.array([[1, 2],
 [2, 3],
 [4, 5]])

deffun(row):
returnnp.array([[1, 1], [3, 3], [4, 4]])

expected_result=np.array([[[1, 1], [1, 1]],
 [[3, 3], [3, 3]],
 [[4, 4], [4, 4]]])

assert (preprocess_tuples(array, fun) ==expected_result).all()


deftest_preprocess_points_simple_example():
"""Test the preprocessor on very simple examples of points to ensure the
 result is as expected"""
array=np.array([1, 2, 4])

deffun(row):
return [[1, 1], [3, 3], [4, 4]]

expected_result=np.array([[1, 1],
 [3, 3],
 [4, 4]])

assert (preprocess_points(array, fun) ==expected_result).all()


@pytest.mark.parametrize('estimator, build_dataset', metric_learners,
ids=ids_metric_learners)
deftest_same_with_or_without_preprocessor(estimator, build_dataset):
"""Test that algorithms using a preprocessor behave consistently
# with their no-preprocessor equivalent
 """
dataset_indices=build_dataset(with_preprocessor=True)
dataset_formed=build_dataset(with_preprocessor=False)
X=dataset_indices.preprocessor
indicators_to_transform=dataset_indices.to_transform
formed_points_to_transform=dataset_formed.to_transform
 (indices_train, indices_test, y_train, y_test, formed_train,
formed_test) =train_test_split(dataset_indices.data,
dataset_indices.target,
dataset_formed.data,
random_state=SEED)

estimator_with_preprocessor=clone(estimator)
set_random_state(estimator_with_preprocessor)
estimator_with_preprocessor.set_params(preprocessor=X)
estimator_with_preprocessor.fit(*remove_y(estimator, indices_train, y_train))

estimator_without_preprocessor=clone(estimator)
set_random_state(estimator_without_preprocessor)
estimator_without_preprocessor.set_params(preprocessor=None)
estimator_without_preprocessor.fit(*remove_y(estimator, formed_train,
y_train))

estimator_with_prep_formed=clone(estimator)
set_random_state(estimator_with_prep_formed)
estimator_with_prep_formed.set_params(preprocessor=X)
estimator_with_prep_formed.fit(*remove_y(estimator, indices_train, y_train))

# test prediction methods
formethodin ["predict", "decision_function"]:
ifhasattr(estimator, method):
output_with_prep=getattr(estimator_with_preprocessor,
method)(indices_test)
output_without_prep=getattr(estimator_without_preprocessor,
method)(formed_test)
assertnp.array(output_with_prep==output_without_prep).all()
output_with_prep=getattr(estimator_with_preprocessor,
method)(indices_test)
output_with_prep_formed=getattr(estimator_with_prep_formed,
method)(formed_test)
assertnp.array(output_with_prep==output_with_prep_formed).all()

# Test pair_score, all learners have it.
idx1=np.array([[0, 2], [5, 3]], dtype=int)
output_with_prep=estimator_with_preprocessor.pair_score(
indicators_to_transform[idx1])
output_without_prep=estimator_without_preprocessor.pair_score(
formed_points_to_transform[idx1])
assertnp.array(output_with_prep==output_without_prep).all()

output_with_prep=estimator_with_preprocessor.pair_score(
indicators_to_transform[idx1])
output_without_prep=estimator_with_prep_formed.pair_score(
formed_points_to_transform[idx1])
assertnp.array(output_with_prep==output_without_prep).all()

# Test pair_distance
not_implemented_msg=""
# Todo in 0.7.0: Change 'not_implemented_msg' for the message that says
# "This learner does not have pair_distance"
try:
output_with_prep=estimator_with_preprocessor.pair_distance(
indicators_to_transform[idx1])
output_without_prep=estimator_without_preprocessor.pair_distance(
formed_points_to_transform[idx1])
assertnp.array(output_with_prep==output_without_prep).all()

output_with_prep=estimator_with_preprocessor.pair_distance(
indicators_to_transform[idx1])
output_without_prep=estimator_with_prep_formed.pair_distance(
formed_points_to_transform[idx1])
assertnp.array(output_with_prep==output_without_prep).all()

exceptExceptionasraised_exception:
assertraised_exception.value.args[0] ==not_implemented_msg

# Test transform
not_implemented_msg=""
# Todo in 0.7.0: Change 'not_implemented_msg' for the message that says
# "This learner does not have transform"
try:
output_with_prep=estimator_with_preprocessor.transform(
indicators_to_transform)
output_without_prep=estimator_without_preprocessor.transform(
formed_points_to_transform)
assertnp.array(output_with_prep==output_without_prep).all()

output_with_prep=estimator_with_preprocessor.transform(
indicators_to_transform)
output_without_prep=estimator_with_prep_formed.transform(
formed_points_to_transform)
assertnp.array(output_with_prep==output_without_prep).all()

exceptExceptionasraised_exception:
assertraised_exception.value.args[0] ==not_implemented_msg


deftest_check_collapsed_pairs_raises_no_error():
"""Checks that check_collapsed_pairs raises no error if no collapsed pairs