1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
"""
test_python.py: Testing libmscp through the mscp python binding.
"""
import pytest
import mscp
import os
from util import File, check_same_md5sum
def test_create_and_release():
m = mscp.mscp("localhost", mscp.LOCAL2REMOTE)
m.cleanup()
""" copy test """
remote = "localhost"
remote_prefix = "{}/".format(os.getcwd()) # use current dir
param_remote_prefix_and_direction = [
("", remote_prefix, mscp.LOCAL2REMOTE), (remote_prefix, "", mscp.REMOTE2LOCAL)
]
param_single_copy = [
(File("src", size = 64), File("dst")),
(File("src", size = 4096 * 1), File("dst")),
(File("src", size = 128 * 1024 * 1024), File("dst")),
]
@pytest.mark.parametrize("src_prefix, dst_prefix, direction",
param_remote_prefix_and_direction)
@pytest.mark.parametrize("src, dst", param_single_copy)
def test_single_copy(src_prefix, dst_prefix, direction, src, dst):
src.make()
m = mscp.mscp(remote, direction)
m.copy(src_prefix + src.path, dst_prefix + dst.path)
assert check_same_md5sum(src, dst)
src.cleanup()
dst.cleanup()
param_double_copy = [
(File("src1", size = 1024 * 1024), File("src2", size = 1024 * 1024),
File("dst/src1"), File("dst/src2")
)
]
@pytest.mark.parametrize("src_prefix, dst_prefix, direction",
param_remote_prefix_and_direction)
@pytest.mark.parametrize("s1, s2, d1, d2", param_double_copy)
def test_double_copy(src_prefix, dst_prefix, direction, s1, s2, d1, d2):
s1.make()
s2.make()
mscp.mscp(remote, direction).copy([src_prefix + s1.path, src_prefix + s2.path],
dst_prefix + "dst")
assert check_same_md5sum(s1, d1)
assert check_same_md5sum(s2, d2)
s1.cleanup()
s2.cleanup()
d1.cleanup()
d2.cleanup()
param_single_copy = [
(File("src", size = 1024 * 1024 * 4), File("dst")),
]
param_kwargs = [
{ "nr_threads": 6 },
{ "nr_ahead": 64 },
{ "min_chunk_sz": 1 * 1024 * 1024 },
{ "max_chunk_sz": 64 * 1024 * 1024 },
{ "coremask": "0x0f" },
{ "max_startups": 5 },
{ "severity": mscp.SEVERITY_NONE },
{ "cipher": "aes128-gcm@openssh.com" },
{ "compress": "yes" },
{ "no_hostkey_check": True },
{ "enable_nagle": True },
]
@pytest.mark.parametrize("src_prefix, dst_prefix, direction",
param_remote_prefix_and_direction)
@pytest.mark.parametrize("src, dst", param_single_copy)
@pytest.mark.parametrize("kw", param_kwargs)
def test_kwargs(src_prefix, dst_prefix, direction, src, dst, kw):
src.make()
m = mscp.mscp(remote, direction, **kw)
m.copy(src_prefix + src.path, dst_prefix + dst.path)
assert check_same_md5sum(src, dst)
src.cleanup()
dst.cleanup()
def test_login_failed():
m = mscp.mscp("asdfasdf@" + remote, mscp.LOCAL2REMOTE)
with pytest.raises(RuntimeError) as e:
m.connect()
m = mscp.mscp(remote, mscp.LOCAL2REMOTE, login_name = "asdfadsf")
with pytest.raises(RuntimeError) as e:
m.connect()
m = mscp.mscp(remote, mscp.LOCAL2REMOTE, port = "65534")
with pytest.raises(RuntimeError) as e:
m.connect()
def test_get_stat_before_copy_start():
m = mscp.mscp("localhost", mscp.LOCAL2REMOTE)
m.connect()
(total, done, finished) = m.stats()
assert total == 0 and done == 0
param_invalid_kwargs = [
{ "nr_threads": -1 },
{ "nr_ahead": -1 },
{ "min_chunk_sz": 1 },
{ "max_chunk_sz": 1 },
{ "coremask": "xxxxx" },
{ "max_startups": -1 },
{ "cipher": "invalid" },
{ "hmac": "invalid"},
{ "compress": "invalid"},
]
@pytest.mark.parametrize("kw", param_invalid_kwargs)
def test_invalid_options(kw):
with pytest.raises(RuntimeError) as e:
m = mscp.mscp("localhost", mscp.LOCAL2REMOTE, **kw)
m.connect()
|
