Raise error for groups that decorate non-operation functions

changelog.txt

@@ -21,6 +21,7 @@ Fixed
 
 - Scripts are now generated correctly when project path contains spaces and special characters (#572).
 - XSEDE Expanse template has been fixed to remove leading spaces (#575, #576).
+- Error raised when a ``FlowGroup`` decorates functions that are not ``FlowProject`` operations (#538, #544).
 
 Changed
 +++++++

flow/project.py

@@ -641,6 +641,8 @@ class FlowGroupEntry:
     ----------
     name : str
         The name of the :class:`FlowGroup` to be created.
+    project : flow.FlowProject
+        The project the group is associated with.
     options : str
         The :meth:`FlowProject.run` options to pass when submitting the group.
         These will be included in all submissions. Submissions use run
@@ -649,8 +651,9 @@ class FlowGroupEntry:
         aggregator object associated with the :class:`FlowGroup` (Default value = None).
     """
 
-    def __init__(self, name, options="", group_aggregator=None):
+    def __init__(self, name, project, options="", group_aggregator=None):
         self.name = name
+        self._project = project
         self.options = options
         # We register aggregators associated with operation functions in
         # `_register_groups` and we do not set the aggregator explicitly.
@@ -675,14 +678,22 @@ def __call__(self, func):
             The decorated function.
 
         """
+        if not any(
+            func == op_func for _, op_func in self._project._OPERATION_FUNCTIONS
+        ):
+            raise FlowProjectDefinitionError(
+                f"Cannot add function '{func}' to group without making the function an "
+                f"operation. Add @MyProjectClass.operation below group decorator."
+            )
+
         if hasattr(func, "_flow_groups"):
             if self.name in func._flow_groups:
                 raise FlowProjectDefinitionError(
                     f"Cannot reregister operation '{func}' with the group '{self.name}'."
                 )
-            func._flow_groups.append(self.name)
+            func._flow_groups.add(self.name)
         else:
-            func._flow_groups = [self.name]
+            func._flow_groups = {self.name}
         return func
 
     def _set_directives(self, func, directives):
@@ -1455,11 +1466,13 @@ def __call__(self, func, name=None):
                 # delay setting the aggregator because we do not restrict the decorator
                 # placement in terms of `@FlowGroupEntry`, `@aggregator`, or
                 # `@operation`.
-                self._parent_class._GROUPS.append(FlowGroupEntry(name=name, options=""))
+                self._parent_class._GROUPS.append(
+                    FlowGroupEntry(name=name, project=self._parent_class, options="")
+                )
                 if hasattr(func, "_flow_groups"):
-                    func._flow_groups.append(name)
+                    func._flow_groups.add(name)
                 else:
-                    func._flow_groups = [name]
+                    func._flow_groups = {name}
                 return func
 
             def with_directives(self, directives, name=None):
@@ -2989,7 +3002,7 @@ def _op_submission_summary(counter):
         num_omitted_operations = len(op_counter) - len(context["op_counter"])
         if num_omitted_operations > 0:
             context["op_counter"].append(
-                (f"[{num_omitted_operations} more operations omitted]", "")
+                (f"[{num_omitted_operations} more operations omitted]", "", "")
             )
 
         # We have to make a deep copy of the template environment if we're
@@ -4299,7 +4312,7 @@ def foo(job):
             )
         cls._GROUP_NAMES.add(name)
         group_entry = FlowGroupEntry(
-            name=name, options=options, group_aggregator=group_aggregator
+            name=name, project=cls, options=options, group_aggregator=group_aggregator
         )
         cls._GROUPS.append(group_entry)
         return group_entry

tests/define_aggregate_test_project.py

@@ -49,15 +49,15 @@ def agg_op1_custom(*jobs):
     set_all_job_docs(jobs, "sum_custom", sum_custom)
 
 
-@_AggregateTestProject.operation
 @group1
+@_AggregateTestProject.operation
 @aggregator.groupsof(30)
 def agg_op2(*jobs):
     set_all_job_docs(jobs, "op2", True)
 
 
-@_AggregateTestProject.operation
 @group1
+@_AggregateTestProject.operation
 @aggregator()
 def agg_op3(*jobs):
     set_all_job_docs(jobs, "op3", True)

tests/define_directives_test_project.py

@@ -10,22 +10,22 @@ class _DirectivesTestProject(FlowProject):
 group = _DirectivesTestProject.make_group(name="walltimegroup")
 
 
-@_DirectivesTestProject.operation.with_directives({"walltime": 1.0})
 @group
+@_DirectivesTestProject.operation.with_directives({"walltime": 1.0})
 def op_walltime(job):
     pass
 
 
-@_DirectivesTestProject.operation.with_directives({"walltime": None})
 @group
+@_DirectivesTestProject.operation.with_directives({"walltime": None})
 def op_walltime_2(job):
     pass
 
 
+@group
 @_DirectivesTestProject.operation.with_directives(
     {"walltime": datetime.timedelta(hours=2)}
 )
-@group
 def op_walltime_3(job):
     pass
 

tests/define_status_test_project.py

@@ -37,21 +37,22 @@ def b_is_even(job):
 
 @flow.cmd
 @group1
+@_TestProject.operation
 @_TestProject.pre(b_is_even)
 @_TestProject.post.isfile("world.txt")
 def op1(job):
     return 'echo "hello" > {job.ws}/world.txt'
 
 
-@_TestProject.operation
 @group1
+@_TestProject.operation
 @_TestProject.post.true("test")
 def op2(job):
     job.document.test = os.getpid()
 
 
-@_TestProject.operation
 @group2
+@_TestProject.operation
 @_TestProject.post.true("test2")
 def op3(job):
     job.document.test2 = True

tests/define_template_test_project.py

@@ -13,14 +13,14 @@ class TestProject(flow.FlowProject):
 group1 = TestProject.make_group(name="group1")
 
 
-@TestProject.operation
 @group1
+@TestProject.operation
 def serial_op(job):
     pass
 
 
-@TestProject.operation.with_directives({"np": TestProject.np})
 @group1
+@TestProject.operation.with_directives({"np": TestProject.np})
 def parallel_op(job):
     pass
 
@@ -56,14 +56,14 @@ def mpi_gpu_op(job):
     pass
 
 
-@TestProject.operation.with_directives({"memory": TestProject.memory})
 @group1
+@TestProject.operation.with_directives({"memory": TestProject.memory})
 def memory_op(job):
     pass
 
 
-@TestProject.operation.with_directives({"walltime": TestProject.walltime})
 @group1
+@TestProject.operation.with_directives({"walltime": TestProject.walltime})
 def walltime_op(job):
     pass
 

tests/define_test_project.py

@@ -35,6 +35,7 @@ def b_is_even(job):
         return False
 
 
+@group1
 @_TestProject.operation.with_directives(
     {
         # Explicitly set a "bad" directive that is unused by the template.
@@ -46,7 +47,6 @@ def b_is_even(job):
 )
 @flow.cmd
 @_TestProject.pre(b_is_even)
-@group1
 @_TestProject.post.isfile("world.txt")
 def op1(job):
     return 'echo "hello" > {job.ws}/world.txt'
@@ -56,15 +56,15 @@ def _need_to_fork(job):
     return job.doc.get("fork")
 
 
-@_TestProject.operation.with_directives({"fork": _need_to_fork})
 @group1
+@_TestProject.operation.with_directives({"fork": _need_to_fork})
 @_TestProject.post.true("test")
 def op2(job):
     job.document.test = os.getpid()
 
 
-@_TestProject.operation.with_directives({"ngpu": 1, "omp_num_threads": 1})
 @group2.with_directives(dict(omp_num_threads=4))
+@_TestProject.operation.with_directives({"ngpu": 1, "omp_num_threads": 1})
 @_TestProject.post.true("test3_true")
 @_TestProject.post.false("test3_false")
 @_TestProject.post.not_(lambda job: job.doc.test3_false)
@@ -80,8 +80,8 @@ class _DynamicTestProject(_TestProject):
 group3 = _DynamicTestProject.make_group(name="group3")
 
 
-@_DynamicTestProject.operation
 @group3
+@_DynamicTestProject.operation
 @_DynamicTestProject.pre.after(op1)
 @_DynamicTestProject.post(lambda job: job.sp.get("dynamic", False))
 def op4(job):

tests/test_project.py

@@ -269,9 +269,9 @@ def baz(job):
             pass
 
         with suspend_logging():
-            a = A.get_project(root=self._tmp_dir.name)
-            b = B.get_project(root=self._tmp_dir.name)
-            c = C.get_project(root=self._tmp_dir.name)
+            a = self.mock_project(A)
+            b = self.mock_project(B)
+            c = self.mock_project(C)
 
         assert len(a.operations) == 2
         assert len(b.operations) == 3
@@ -288,7 +288,23 @@ class A(FlowProject):
             def op1(job):
                 pass
 
-        return
+    def test_repeat_anonymous_operation_definition(self):
+        class A(FlowProject):
+            pass
+
+        A.operation(lambda job: print("Hello", job))
+
+        assert len(self.mock_project(A).operations) == 1
+
+        anonymous_func = lambda job: print("Hi", job)  # noqa: E731
+
+        with pytest.raises(FlowProjectDefinitionError):
+            # Only one anonymous operation is allowed, or else the name
+            # "<lambda>" conflicts between the operations.
+            A.operation(anonymous_func)
+
+        A.operation(anonymous_func, name="hi_operation")
+        assert len(self.mock_project(A).operations) == 2
 
     def test_repeat_operation_name(self):
         class A(FlowProject):
@@ -347,10 +363,10 @@ def op1(job):
 
         # Should raise no error
         with suspend_logging():
-            A.get_project(root=self._tmp_dir.name)
+            self.mock_project(A)
 
         with pytest.raises(FlowProjectDefinitionError):
-            B.get_project(root=self._tmp_dir.name)
+            self.mock_project(B)
 
     def test_label_definition(self):
         class A(FlowProject):
@@ -371,9 +387,9 @@ def label1(job):
         def label2(job):
             pass
 
-        a = A.get_project(root=self._tmp_dir.name)
-        b = B.get_project(root=self._tmp_dir.name)
-        c = C.get_project(root=self._tmp_dir.name)
+        a = self.mock_project(A)
+        b = self.mock_project(B)
+        c = self.mock_project(C)
 
         assert len(a._label_functions) == 1
         assert len(b._label_functions) == 2
@@ -1772,6 +1788,37 @@ def bar(job):
         with pytest.raises(FlowProjectDefinitionError):
             B.make_group("bar")
 
+    def test_group_operation_without_operation_definition(self):
+        """Test that groups can only be applied to operations."""
+
+        class A(FlowProject):
+            pass
+
+        group = A.make_group("foo")
+
+        with pytest.raises(FlowProjectDefinitionError):
+
+            @group
+            def test_op(job):
+                pass
+
+        # Make test_op into an operation, then project creation should succeed.
+        @group
+        @A.operation
+        def test_op_2(job):
+            pass
+
+    def test_group_operation_without_operation_definition_anonymous(self):
+        """Test that groups cannot be applied to anonymous functions."""
+
+        class A(FlowProject):
+            pass
+
+        group = A.make_group("foo")
+
+        with pytest.raises(FlowProjectDefinitionError):
+            group(lambda job: print(job))
+
     def test_repeat_group_definition(self):
         """Test that groups cannot be registered if a group with that name exists."""