16470: Adds ArvadosModel test exposing a bug with audit logs.
[arvados.git] / services / fuse / tests / test_tmp_collection.py
1 # Copyright (C) The Arvados Authors. All rights reserved.
2 #
3 # SPDX-License-Identifier: AGPL-3.0
4
5 from builtins import range
6 from six import assertRegex
7 import arvados
8 import arvados_fuse
9 import arvados_fuse.command
10 import json
11 import logging
12 import os
13 import tempfile
14 import unittest
15
16 from .integration_test import IntegrationTest
17 from .mount_test_base import MountTestBase
18
19 logger = logging.getLogger('arvados.arv-mount')
20
21
22 class TmpCollectionArgsTest(unittest.TestCase):
23     def setUp(self):
24         self.tmpdir = tempfile.mkdtemp()
25
26     def tearDown(self):
27         os.rmdir(self.tmpdir)
28
29     def test_tmp_only(self):
30         args = arvados_fuse.command.ArgumentParser().parse_args([
31             '--mount-tmp', 'tmp1',
32             '--mount-tmp', 'tmp2',
33             self.tmpdir,
34         ])
35         self.assertIn(args.mode, [None, 'custom'])
36         self.assertEqual(['tmp1', 'tmp2'], args.mount_tmp)
37         for mtype in ['home', 'shared', 'by_id', 'by_pdh', 'by_tag']:
38             self.assertEqual([], getattr(args, 'mount_'+mtype))
39
40     def test_tmp_and_home(self):
41         args = arvados_fuse.command.ArgumentParser().parse_args([
42             '--mount-tmp', 'test_tmp',
43             '--mount-home', 'test_home',
44             self.tmpdir,
45         ])
46         self.assertIn(args.mode, [None, 'custom'])
47         self.assertEqual(['test_tmp'], args.mount_tmp)
48         self.assertEqual(['test_home'], args.mount_home)
49
50     def test_no_tmp(self):
51         args = arvados_fuse.command.ArgumentParser().parse_args([
52             self.tmpdir,
53         ])
54         self.assertEqual([], args.mount_tmp)
55
56
57 def current_manifest(tmpdir):
58     with open(os.path.join(tmpdir, '.arvados#collection')) as tmp:
59         return json.load(tmp)['manifest_text']
60
61
62 class TmpCollectionTest(IntegrationTest):
63     mnt_args = [
64         '--read-write',
65         '--mount-tmp', 'zzz',
66     ]
67
68     @IntegrationTest.mount(argv=mnt_args+['--mount-tmp', 'yyy'])
69     def test_two_tmp(self):
70         self.pool_test(os.path.join(self.mnt, 'zzz'),
71                        os.path.join(self.mnt, 'yyy'))
72     @staticmethod
73     def _test_two_tmp(self, zzz, yyy):
74         self.assertEqual(current_manifest(zzz), "")
75         self.assertEqual(current_manifest(yyy), "")
76         with open(os.path.join(zzz, 'foo'), 'w') as f:
77             f.write('foo')
78         self.assertNotEqual(current_manifest(zzz), "")
79         self.assertEqual(current_manifest(yyy), "")
80         os.unlink(os.path.join(zzz, 'foo'))
81         with open(os.path.join(yyy, 'bar'), 'w') as f:
82             f.write('bar')
83         self.assertEqual(current_manifest(zzz), "")
84         self.assertNotEqual(current_manifest(yyy), "")
85
86     @IntegrationTest.mount(argv=mnt_args)
87     def test_tmp_empty(self):
88         self.pool_test(os.path.join(self.mnt, 'zzz'))
89     @staticmethod
90     def _test_tmp_empty(self, tmpdir):
91         self.assertEqual(current_manifest(tmpdir), "")
92
93     @IntegrationTest.mount(argv=mnt_args)
94     def test_tmp_onefile(self):
95         self.pool_test(os.path.join(self.mnt, 'zzz'))
96     @staticmethod
97     def _test_tmp_onefile(self, tmpdir):
98         with open(os.path.join(tmpdir, 'foo'), 'w') as f:
99             f.write('foo')
100         assertRegex(
101             self,
102             current_manifest(tmpdir),
103             r'^\. acbd18db4cc2f85cedef654fccc4a4d8\+3(\+\S+)? 0:3:foo\n$')
104
105     @IntegrationTest.mount(argv=mnt_args)
106     def test_tmp_snapshots(self):
107         self.pool_test(os.path.join(self.mnt, 'zzz'))
108     @staticmethod
109     def _test_tmp_snapshots(self, tmpdir):
110         ops = [
111             ('foo', 'bar',
112              r'^\. 37b51d194a7513e45b56f6524f2d51f2\+3(\+\S+)? 0:3:foo\n$'),
113             ('foo', 'foo',
114              r'^\. acbd18db4cc2f85cedef654fccc4a4d8\+3(\+\S+)? 0:3:foo\n$'),
115             ('bar', 'bar',
116              r'^\. 37b51d194a7513e45b56f6524f2d51f2\+3(\+\S+)? acbd18db4cc2f85cedef654fccc4a4d8\+3(\+\S+)? 0:3:bar 3:3:foo\n$'),
117             ('foo', None,
118              r'^\. 37b51d194a7513e45b56f6524f2d51f2\+3(\+\S+)? 0:3:bar\n$'),
119             ('bar', None,
120              r'^$'),
121         ]
122         for _ in range(10):
123             for fn, content, expect in ops:
124                 path = os.path.join(tmpdir, fn)
125                 if content is None:
126                     os.unlink(path)
127                 else:
128                     with open(path, 'w') as f:
129                         f.write(content)
130                 assertRegex(self, current_manifest(tmpdir), expect)
131
132     @IntegrationTest.mount(argv=mnt_args)
133     def test_tmp_rewrite(self):
134         self.pool_test(os.path.join(self.mnt, 'zzz'))
135     @staticmethod
136     def _test_tmp_rewrite(self, tmpdir):
137         with open(os.path.join(tmpdir, "b1"), 'w') as f:
138             f.write("b1")
139         with open(os.path.join(tmpdir, "b2"), 'w') as f:
140             f.write("b2")
141         with open(os.path.join(tmpdir, "b1"), 'w') as f:
142             f.write("1b")
143         assertRegex(self, current_manifest(tmpdir), "^\. ed4f3f67c70b02b29c50ce1ea26666bd\+4(\+\S+)? 0:2:b1 2:2:b2\n$")