auto-testing/test_rustfs_replication.py at main · rustfs/auto-testing · GitHub

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
#!/usr/bin/env python3
"""Run the real replicated-delete case without a cluster."""

import os
import subprocess
import unittest
from pathlib import Path


ROOT = Path(__file__).resolve().parent


class ReplicatedDeleteTests(unittest.TestCase):
    def run_case(self, mode):
        source = (ROOT / "rustfs-replication-test.sh").read_text()
        case = source[source.index("test_rep110() {"):source.index("\ntest_rep111() {")]
        harness = r'''
set -u
SITE2_ENDPOINT=http://target:19000
TEST_TMP=/unused
new_versioned_bucket() { :; }
register_replication_target() { echo arn:test; }
put_replication_rule() { :; }
make_test_file() { :; }
rm_bucket_force() { :; }
aws_wait_ep() { [ "$1" = "$SITE2_ENDPOINT" ]; }
assert_aws_ok() { [ "$AWS_CODE" -eq 0 ]; }
aws_run() {
  case "$2" in
    put-object|delete-object) AWS_CODE=0; AWS_OUT='{}' ;;
    *) AWS_CODE=254; AWS_OUT='NoSuchBucket (source endpoint)' ;;
  esac
}
aws_run_site2() {
  case "$2" in
    head-object)
      AWS_CODE=254
      AWS_OUT='Not Found'
      [ "$CASE_MODE" != head_error ] || AWS_OUT='ServiceUnavailable'
      ;;
    list-objects-v2)
      AWS_CODE=0
      AWS_OUT='{}'
      [ "$CASE_MODE" != listed ] || AWS_OUT='{"Contents":[{"Key":"gone.txt"}]}'
      [ "$CASE_MODE" != list_error ] || AWS_CODE=254
      ;;
    *) return 99 ;;
  esac
  return 0
}
'''
        return subprocess.run(["bash", "-c", harness + case + "\ntest_rep110\n"],
                              env={**os.environ, "CASE_MODE": mode}, capture_output=True,
                              text=True, timeout=5)

    def test_checks_the_target_endpoint(self):
        result = self.run_case("deleted")
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)

    def test_rejects_remaining_object(self):
        self.assertNotEqual(self.run_case("listed").returncode, 0)

    def test_rejects_head_failure_other_than_not_found(self):
        self.assertNotEqual(self.run_case("head_error").returncode, 0)

    def test_rejects_failed_listing_even_with_empty_json(self):
        self.assertNotEqual(self.run_case("list_error").returncode, 0)


class JqBatchCompatibilityTests(unittest.TestCase):
    def test_nwise_batches_without_jq_private_builtin(self):
        helper = ROOT / "scripts/lib/jq_nwise.jq"
        result = subprocess.run(
            ["jq", "-nc", "-L", str(helper.parent),
             'include "jq_nwise"; [1,2,3,4,5] | nwise(2)'],
            capture_output=True, text=True, timeout=10,
        )
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.stdout.strip(), "[[1,2],[3,4],[5]]")

    def test_nwise_callers_stream_each_delete_batch(self):
        helper = ROOT / "scripts/lib/jq_nwise.jq"
        result = subprocess.run(
            ["jq", "-nc", "-L", str(helper.parent),
             'include "jq_nwise"; [1,2,3,4,5] | nwise(2)[] | {Objects: .}'],
            capture_output=True, text=True, timeout=10,
        )
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(
            result.stdout.splitlines(),
            ['{"Objects":[1,2]}', '{"Objects":[3,4]}', '{"Objects":[5]}'],
        )


if __name__ == "__main__":
    unittest.main()