Fixed PublishEventBatch diagram in VES spec
[vnfrqts/requirements.git] / fix_invalid_metadata.py
1 # -*- coding: utf8 -*-
2 # org.onap.vnfrqts/requirements
3 # ============LICENSE_START====================================================
4 # Copyright © 2018 AT&T Intellectual Property. All rights reserved.
5 #
6 # Unless otherwise specified, all software contained herein is licensed
7 # under the Apache License, Version 2.0 (the "License");
8 # you may not use this software except in compliance with the License.
9 # You may obtain a copy of the License at
10 #
11 #             http://www.apache.org/licenses/LICENSE-2.0
12 #
13 # Unless required by applicable law or agreed to in writing, software
14 # distributed under the License is distributed on an "AS IS" BASIS,
15 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 # See the License for the specific language governing permissions and
17 # limitations under the License.
18 #
19 # Unless otherwise specified, all documentation contained herein is licensed
20 # under the Creative Commons License, Attribution 4.0 Intl. (the "License");
21 # you may not use this documentation except in compliance with the License.
22 # You may obtain a copy of the License at
23 #
24 #             https://creativecommons.org/licenses/by/4.0/
25 #
26 # Unless required by applicable law or agreed to in writing, documentation
27 # distributed under the License is distributed on an "AS IS" BASIS,
28 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
29 # See the License for the specific language governing permissions and
30 # limitations under the License.
31 #
32 # ============LICENSE_END============================================
33
34 """
35 This script will consume the `invalid_metadata.csv` file produced by
36 `gen_requirement_changes.py`, then add/update any `:introduced:` or `:updated:`
37 attributes that may be missing from req directives.
38 """
39 import csv
40 import os
41 import re
42 from collections import OrderedDict
43
44 import pytest
45
46 INPUT_FILE = "invalid_metadata.csv"
47
48
49 def load_invalid_reqs(fileobj):
50     """Load the invalid requirements from the input file into a dict"""
51     reader = csv.reader(fileobj)
52     next(reader)  # skip header
53     return {row[0]: (row[1].strip(), row[2].strip()) for row in reader}
54
55
56 def check(predicate, msg):
57     """Raises a RuntimeError with the given msg if the predicate is false"""
58     if not predicate:
59         raise RuntimeError(msg)
60
61
62 class MetadataFixer:
63     """Takes a dict of requirement ID to expected metadata value.  The
64     NeedsVisitor will pass the requirement attributes as a a dict
65     to `__call__`.  If the requirement is one that needs to be fixed, then
66     it will add or update the attributes as needed and return it to the
67     visitor, otherwise it will return the attributes unchanged."""
68
69     def __init__(self, reqs_to_fix):
70         """Initialize the fixer with a dict of requirement ID to tuple of
71         (attr name, attr value)."""
72         self.reqs_to_fix = reqs_to_fix
73
74     def __call__(self, metadata):
75         """If metadata is for a requirement that needs to be fixed, then
76         adds or updates the attribute as needed and returns it, otherwise
77         it returns metadata unchanged."""
78         r_id = metadata[":id:"]
79         if r_id in self.reqs_to_fix:
80             attr, value = self.reqs_to_fix[r_id]
81             metadata[attr] = value
82         return metadata
83
84
85 class NeedsVisitor:
86     """Walks a directory for reStructuredText files and detects needs
87     directives as defined by sphinxcontrib-needs.  When a directive is
88     found, then attributes are passed to a callback for processing if the
89     callback returns a dict of attributes, then the revised dict is used
90     instead of the attributes that were passed"""
91
92     def __init__(self, func):
93         self.directives = re.compile("\.\.\s+req::.*")
94         self.func = func
95
96     def process(self, root_dir):
97         """Walks the `root_dir` looking for rst to files to parse"""
98         for dir_path, sub_dirs, filenames in os.walk(root_dir):
99             for filename in filenames:
100                 if filename.lower().endswith(".rst"):
101                     self.handle_rst_file(os.path.join(dir_path, filename))
102
103     @staticmethod
104     def read(path):
105         """Read file at `path` and return lines as list"""
106         with open(path, "r") as f:
107             print("path=", path)
108             return list(f)
109
110     @staticmethod
111     def write(path, content):
112         """Write a content to the given path"""
113         with open(path, "w") as f:
114             for line in content:
115                 f.write(line)
116
117     def handle_rst_file(self, path):
118         lines = (line for line in self.read(path))
119         new_contents = []
120         for line in lines:
121             if self.is_needs_directive(line):
122                 metadata_lines = self.handle_need(lines)
123                 new_contents.append(line)
124                 new_contents.extend(metadata_lines)
125             else:
126                 new_contents.append(line)
127         self.write(path, new_contents)
128
129     def is_needs_directive(self, line):
130         """Returns True if the line denotes the start of a needs directive"""
131         return bool(self.directives.match(line))
132
133     def handle_need(self, lines):
134         """Called when a needs directive is encountered.  The lines
135         will be positioned on the line after the directive.  The attributes
136         will be read, and then passed to the visitor for processing"""
137         attributes = OrderedDict()
138         indent = 4
139         for line in lines:
140             if line.strip().startswith(":"):
141                 indent = self.calc_indent(line)
142                 attr, value = self.parse_attribute(line)
143                 attributes[attr] = value
144             else:
145                 if attributes:
146                     new_attributes = self.func(attributes)
147                     attr_lines = self.format_attributes(new_attributes, indent)
148                     return attr_lines + [line]
149                 else:
150                     return [line]
151
152     @staticmethod
153     def format_attributes(attrs, indent):
154         """Converts a dict back to properly indented lines"""
155         spaces = " " * indent
156         return ["{}{} {}\n".format(spaces, k, v) for k, v in attrs.items()]
157
158     @staticmethod
159     def parse_attribute(line):
160         return re.split("\s+", line.strip(), maxsplit=1)
161
162     @staticmethod
163     def calc_indent(line):
164         return len(line) - len(line.lstrip())
165
166
167 if __name__ == '__main__':
168     with open(INPUT_FILE, "r") as f:
169         invalid_reqs = load_invalid_reqs(f)
170     metadata_fixer = MetadataFixer(invalid_reqs)
171     visitor = NeedsVisitor(metadata_fixer)
172     visitor.process("docs")
173
174
175 # Tests
176 @pytest.fixture
177 def metadata_fixer():
178     fixes = {
179         "R-1234": (":introduced:", "casablanca"),
180         "R-5678": (":updated:", "casablanca"),
181     }
182     return MetadataFixer(fixes)
183
184
185 def test_check_raises_when_false():
186     with pytest.raises(RuntimeError):
187         check(False, "error")
188
189
190 def test_check_does_not_raise_when_true():
191     check(True, "error")
192
193
194 def test_load_invalid_req():
195     contents = [
196         "reqt_id, attribute, value",
197         "R-1234,:introduced:, casablanca",
198         "R-5678,:updated:, casablanca",
199     ]
200     result = load_invalid_reqs(contents)
201     assert len(result) == 2
202     assert result["R-1234"][0] == ":introduced:"
203     assert result["R-1234"][1] == "casablanca"
204
205
206 def test_metadata_fixer_adds_when_missing(metadata_fixer):
207     attributes = {":id:": "R-5678", ":introduced:": "beijing"}
208     result = metadata_fixer(attributes)
209     assert ":updated:" in result
210     assert result[":updated:"] == "casablanca"
211
212
213 def test_metadata_fixer_updates_when_incorrect(metadata_fixer):
214     attributes = {":id:": "R-5678", ":updated:": "beijing"}
215     result = metadata_fixer(attributes)
216     assert ":updated:" in result
217     assert result[":updated:"] == "casablanca"
218     assert ":introduced:" not in result
219
220
221 def test_needs_visitor_process(monkeypatch):
222     v = NeedsVisitor(lambda x: x)
223     paths = []
224
225     def mock_handle_rst(path):
226         paths.append(path)
227
228     monkeypatch.setattr(v, "handle_rst_file", mock_handle_rst)
229     v.process("docs")
230
231     assert len(paths) > 1
232     assert all(path.endswith(".rst") for path in paths)
233
234
235 def test_needs_visitor_is_needs_directive():
236     v = NeedsVisitor(lambda x: x)
237     assert v.is_needs_directive(".. req::")
238     assert not v.is_needs_directive("test")
239     assert not v.is_needs_directive(".. code::")
240
241
242 def test_needs_visitor_format_attributes():
243     v = NeedsVisitor(lambda x: x)
244     attr = OrderedDict()
245     attr[":id:"] = "R-12345"
246     attr[":updated:"] = "casablanca"
247     lines = v.format_attributes(attr, 4)
248     assert len(lines) == 2
249     assert lines[0] == "    :id: R-12345"
250     assert lines[1] == "    :updated: casablanca"
251
252
253 def test_needs_visitor_parse_attributes():
254     v = NeedsVisitor(lambda x: x)
255     assert v.parse_attribute("   :id: R-12345") == [":id:", "R-12345"]
256     assert v.parse_attribute("   :key: one two") == [":key:", "one two"]
257
258
259 def test_needs_visitor_calc_indent():
260     v = NeedsVisitor(lambda x: x)
261     assert v.calc_indent("    test") == 4
262     assert v.calc_indent("   test") == 3
263     assert v.calc_indent("test") == 0
264
265
266 def test_needs_visitor_no_change(monkeypatch):
267     v = NeedsVisitor(lambda x: x)
268     lines = """.. req::
269         :id: R-12345
270         :updated: casablanca
271         
272         Here's my requirement"""
273     monkeypatch.setattr(v, "read", lambda path: lines.split("\n"))
274     result = []
275     monkeypatch.setattr(v, "write", lambda _, content: result.extend(content))
276
277     v.handle_rst_file("dummy_path")
278     assert len(result) == 5
279     assert "\n".join(result) == lines
280
281
282 def test_needs_visitor_with_fix(monkeypatch):
283     fixer = MetadataFixer({"R-12345": (":introduced:", "casablanca")})
284     v = NeedsVisitor(fixer)
285     lines = """.. req::
286         :id: R-12345
287
288         Here's my requirement"""
289     monkeypatch.setattr(v, "read", lambda path: lines.split("\n"))
290     result = []
291     monkeypatch.setattr(v, "write", lambda _, content: result.extend(content))
292
293     v.handle_rst_file("dummy_path")
294     assert len(result) == 5
295     assert ":introduced: casablanca" in "\n".join(result)
296
297
298 def test_load_invalid_reqs():
299     input_file = [
300         "r_id,attr,value",
301         "R-12345,:updated:,casablanca"
302     ]
303     result = load_invalid_reqs(input_file)
304     assert "R-12345" in result
305     assert result["R-12345"][0] == ":updated:"
306     assert result["R-12345"][1] == "casablanca"