Optional task exclusion doesn't function correctly
Executing `./build.py -T -X -o publish` we get:

```
Tasks that will be executed for project "PyBuilder":
    publish - Publishes the project.
              depends on tasks: package
```

with the following output

```
[DEBUG] Registering initializer 'initialize'
[DEBUG] Skipping resolution for excluded task 'clean'
[DEBUG] Skipping resolution for excluded task 'compile_sources'
[DEBUG] Skipping resolution for excluded task 'install'
[DEBUG] Skipping resolution for excluded task 'package'
[DEBUG] Skipping resolution for excluded task 'prepare'
[DEBUG] Skipping resolution for excluded task 'print_module_path'
[DEBUG] Skipping resolution for excluded task 'print_scripts_path'
[DEBUG] Skipping resolution for excluded task 'publish'
[DEBUG] Skipping resolution for excluded task 'run_integration_tests'
[DEBUG] Skipping resolution for excluded task 'run_unit_tests'
[DEBUG] Skipping resolution for excluded task 'verify'
[DEBUG] Skipping resolution for excluded task 'pytddmon'
[DEBUG] Skipping resolution for excluded task 'upload'
[DEBUG] Skipping resolution for excluded task 'install_build_dependencies'
[DEBUG] Skipping resolution for excluded task 'install_dependencies'
[DEBUG] Skipping resolution for excluded task 'install_runtime_dependencies'
[DEBUG] Skipping resolution for excluded task 'list_dependencies'
[DEBUG] Skipping resolution for excluded task 'build_source_distribution'
[DEBUG] Skipping resolution for excluded task 'run_cram_tests'
[DEBUG] Skipping resolution for excluded task 'analyze'
[DEBUG] Skipping resolution for excluded task 'sphinx_generate_documentation'
[DEBUG] Skipping resolution for excluded task 'sphinx_quickstart'
[DEBUG] Skipping resolution for excluded task 'render_snakefood_report'
[DEBUG] Skipping resolution for excluded task 'pydev_generate'
[DEBUG] Skipping resolution for excluded task 'pycharm_generate'
[DEBUG] Skipping resolution for excluded task 'pdoc_generate_documentation'
```

This is wrong as resolution skipped whole branches that should not have been skipped. That is `publish` does have a hard requirement on `package` which has a hard requirement on `compile_sources` and so on.

The resolution code should be restored and optional tasks should be excluded during the execution path being built.

My apologies for a screw up.

diff --git a/setup.py b/setup.py
index a12fdda..8925ca0 100644
--- a/setup.py
+++ b/setup.py
@@ -36,8 +36,7 @@ script_dir = os.path.dirname(os.path.realpath(__file__))
 build_script = os.path.join(script_dir, "build.py")
 exit_code = 0
 try:
-    subprocess.check_call([build_script, "clean", "install_dependencies", "publish", "-x", "verify", "-x", "analyze",
-                           "-x", "run_unit_tests", "-x", "run_integration_tests", "-x", "coverage"])
+    subprocess.check_call([build_script, "clean", "install_dependencies", "package", "-o"])
     dist_dir = glob.glob(os.path.join(script_dir, "target", "dist", "*"))[0]
     for src_file in glob.glob(os.path.join(dist_dir, "*")):
         file_name = os.path.basename(src_file)
diff --git a/src/integrationtest/python/should_raise_exception_optional_exclude_required_task_tests.py b/src/integrationtest/python/should_raise_exception_optional_exclude_required_task_tests.py
index 7ca4016..dd0a556 100644
--- a/src/integrationtest/python/should_raise_exception_optional_exclude_required_task_tests.py
+++ b/src/integrationtest/python/should_raise_exception_optional_exclude_required_task_tests.py
@@ -42,8 +42,8 @@ def task_c(project):
     project.set_property("c", True)
         """)
         reactor = self.prepare_reactor()
-        self.assertRaises(RequiredTaskExclusionException, reactor.execution_manager.resolve_dependencies,
-                          exclude_optional_tasks=["task_b"])
+        reactor.execution_manager.resolve_dependencies(exclude_optional_tasks=["task_b"])
+        self.assertRaises(RequiredTaskExclusionException, reactor.execution_manager.build_execution_plan, "task_c")
 
 
 if __name__ == "__main__":
diff --git a/src/main/python/pybuilder/cli.py b/src/main/python/pybuilder/cli.py
index d5f82da..b6ea9d2 100644
--- a/src/main/python/pybuilder/cli.py
+++ b/src/main/python/pybuilder/cli.py
@@ -307,7 +307,7 @@ def print_task_list(tasks, quiet=False):
         if task.dependencies:
             whitespace = (column_length + 3) * " "
             depends_on_message = "depends on tasks: %s" % " ".join(
-                task.dependencies)
+                [str(dependency) for dependency in task.dependencies])
             print_text_line(whitespace + depends_on_message)
 
 
diff --git a/src/main/python/pybuilder/execution.py b/src/main/python/pybuilder/execution.py
index 8610698..46f6dee 100644
--- a/src/main/python/pybuilder/execution.py
+++ b/src/main/python/pybuilder/execution.py
@@ -23,10 +23,10 @@
     order regarding dependencies.
 """
 
-import inspect
 import copy
-import traceback
+import inspect
 import sys
+import traceback
 
 import re
 import types
@@ -35,11 +35,11 @@ from pybuilder.errors import (CircularTaskDependencyException,
                               DependenciesNotResolvedException,
                               InvalidNameException,
                               MissingTaskDependencyException,
-                              RequiredTaskExclusionException,
                               MissingActionDependencyException,
-                              NoSuchTaskException)
-from pybuilder.utils import as_list, Timer, odict
+                              NoSuchTaskException,
+                              RequiredTaskExclusionException)
 from pybuilder.graph_utils import Graph, GraphHasCycles
+from pybuilder.utils import as_list, Timer, odict
 
 if sys.version_info[0] < 3:  # if major is less than 3
     from .excp_util_2 import raise_exception
@@ -47,13 +47,19 @@ else:
     from .excp_util_3 import raise_exception
 
 
+def as_task_name(item):
+    if isinstance(item, types.FunctionType):
+        return item.__name__
+    elif hasattr(item, "name"):
+        return item.name
+    else:
+        return str(item)
+
+
 def as_task_name_list(mixed):
     result = []
     for item in as_list(mixed):
-        if isinstance(item, types.FunctionType):
-            result.append(item.__name__)
-        else:
-            result.append(str(item))
+        result.append(as_task_name(item))
     return result
 
 
@@ -100,12 +106,37 @@ class Action(Executable):
         self.teardown = teardown
 
 
+class TaskDependency(object):
+    def __init__(self, mixed, optional=False):
+        self._name = as_task_name(mixed)
+        self._task = mixed if hasattr(mixed, "name") else None
+        self._optional = optional
+
+    def __repr__(self):
+        return self._name if not self._optional else self._name + "(optional)"
+
+    def __eq__(self, other):
+        if isinstance(other, TaskDependency):
+            return self._name == other._name and self._optional == other._optional
+
+    @property
+    def name(self):
+        return self._name
+
+    @property
+    def task(self):
+        return self._task
+
+    @property
+    def optional(self):
+        return self._optional
+
+
 class Task(object):
-    def __init__(self, name, callable, dependencies=None, description="", optional_dependencies=None):
+    def __init__(self, name, callable, dependencies=None, description=""):
         self.name = name
         self.executables = [Executable(name, callable, description)]
-        self.dependencies = as_task_name_list(dependencies)
-        self.optional_dependencies = as_task_name_list(optional_dependencies)
+        self.dependencies = as_list(dependencies)
         self.description = [description]
 
     def __eq__(self, other):
@@ -303,11 +334,13 @@ class ExecutionManager(object):
         return summaries
 
     def get_task(self, name):
+        name = name.name if isinstance(name, TaskDependency) else name
         if not self.has_task(name):
             raise NoSuchTaskException(name)
         return self._tasks[name]
 
     def has_task(self, name):
+        name = name.name if isinstance(name, TaskDependency) else name
         return name in self._tasks
 
     def _collect_transitive_tasks(self, task, visited=None):
@@ -316,12 +349,14 @@ class ExecutionManager(object):
         if task in visited:
             return visited
         visited.add(task)
-        dependencies = [self.get_task(dependency_name) for dependency_name in task.dependencies]
+        dependencies = [dependency for dependency in self._task_dependencies[task.name]]
         for dependency in dependencies:
-            self._collect_transitive_tasks(dependency, visited)
+            self._collect_transitive_tasks(dependency.task, visited)
         return visited
 
     def collect_all_transitive_tasks(self, task_names):
+        self.assert_dependencies_resolved()
+
         all_tasks = set()
         for task_name in task_names:
             all_tasks.update(self._collect_transitive_tasks(self.get_task(task_name)))
@@ -334,14 +369,14 @@ class ExecutionManager(object):
 
         dependency_edges = {}
         for task in self.collect_all_transitive_tasks(as_list(task_names)):
-            dependency_edges[task.name] = task.dependencies
+            dependency_edges[task.name] = [dependency.name for dependency in task.dependencies]
         try:
             Graph(dependency_edges).assert_no_cycles_present()
         except GraphHasCycles as cycles:
             raise CircularTaskDependencyException(str(cycles))
 
         for task_name in as_list(task_names):
-            self.enqueue_task(execution_plan, task_name)
+            self._enqueue_task(execution_plan, task_name)
         return execution_plan
 
     def build_shortest_execution_plan(self, task_names):
@@ -365,17 +400,33 @@ class ExecutionManager(object):
                                                   (self._current_task, task_names, shortest_plan))
         return shortest_plan
 
-    def enqueue_task(self, execution_plan, task_name):
+    def _enqueue_task(self, execution_plan, task_name):
         task = self.get_task(task_name)
 
         if task in execution_plan:
             return
 
         for dependency in self._task_dependencies[task.name]:
-            self.enqueue_task(execution_plan, dependency.name)
+            if self._should_omit_dependency(task, dependency):
+                continue
+            self._enqueue_task(execution_plan, dependency.name)
 
         execution_plan.append(task)
 
+    def _should_omit_dependency(self, task, dependency):
+        if dependency.optional:
+            if self._exclude_all_optional or dependency.name in self._exclude_optional_tasks or \\
+                    dependency.name in self._exclude_tasks:
+                self.logger.debug("Omitting optional dependency '%s' of task '%s'", dependency.name, task.name)
+                return True
+        else:
+            if dependency.name in self._exclude_optional_tasks:
+                raise RequiredTaskExclusionException(task.name, dependency.name)
+            if dependency.name in self._exclude_tasks:
+                self.logger.warn("Omitting required dependency '%s' of task '%s'", dependency.name, task.name)
+                return True
+        return False
+
     def resolve_dependencies(self, exclude_optional_tasks=None, exclude_tasks=None, exclude_all_optional=False):
         self._exclude_optional_tasks = as_task_name_list(exclude_optional_tasks or [])
         self._exclude_tasks = as_task_name_list(exclude_tasks or [])
@@ -385,24 +436,27 @@ class ExecutionManager(object):
             self._execute_before[task.name] = []
             self._execute_after[task.name] = []
             self._task_dependencies[task.name] = []
-            if self.is_task_excluded(task.name) or self.is_optional_task_excluded(task.name):
-                self.logger.debug("Skipping resolution for excluded task '%s'", task.name)
-                continue
-            for d in task.dependencies:
-                if not self.has_task(d):
-                    raise MissingTaskDependencyException(task.name, d)
-                if self.is_optional_task_excluded(d):
-                    raise RequiredTaskExclusionException(task.name, d)
-                if not self.is_task_excluded(d):
-                    self._task_dependencies[task.name].append(self.get_task(d))
-                    self.logger.debug("Adding '%s' as a required dependency of task '%s'", d, task.name)
 
-            for d in task.optional_dependencies:
+            for d in task.dependencies:
+                add_dependency = True
                 if not self.has_task(d):
                     raise MissingTaskDependencyException(task.name, d)
-                if not (self.is_task_excluded(d) or self.is_optional_task_excluded(d)):
-                    self._task_dependencies[task.name].append(self.get_task(d))
-                    self.logger.debug("Adding '%s' as an optional dependency of task '%s'", d, task.name)
+                task_dependencies = self._task_dependencies[task.name]
+                for index, existing_dependency in enumerate(task_dependencies):
+                    if existing_dependency.name == d.name:
+                        if existing_dependency.optional != d.optional:
+                            if existing_dependency.optional:
+                                task_dependencies[index] = TaskDependency(existing_dependency.name)
+                                self.logger.debug("Converting optional dependency '%s' of task '%s' into required",
+                                                  existing_dependency, task.name)
+                            else:
+                                self.logger.debug(
+                                    "Ignoring '%s' as optional dependency of task '%s' - already required",
+                                    existing_dependency, task.name)
+                        add_dependency = False
+                if add_dependency:
+                    self._task_dependencies[task.name].append(TaskDependency(self.get_task(d), d.optional))
+                    self.logger.debug("Adding '%s' as a dependency of task '%s'", d, task.name)
 
         for action in self._actions.values():
             for task in action.execute_before:
@@ -418,9 +472,3 @@ class ExecutionManager(object):
                 self.logger.debug("Adding after action '%s' for task '%s'", action.name, task)
 
         self._dependencies_resolved = True
-
-    def is_task_excluded(self, task):
-        return task in self._exclude_tasks
-
-    def is_optional_task_excluded(self, task):
-        return self._exclude_all_optional or task in self._exclude_optional_tasks
diff --git a/src/main/python/pybuilder/reactor.py b/src/main/python/pybuilder/reactor.py
index ff4d23c..91ce5ae 100644
--- a/src/main/python/pybuilder/reactor.py
+++ b/src/main/python/pybuilder/reactor.py
@@ -32,7 +32,7 @@ from pybuilder.core import (TASK_ATTRIBUTE, DEPENDS_ATTRIBUTE, DEPENDENTS_ATTRIB
                             ACTION_ATTRIBUTE, ONLY_ONCE_ATTRIBUTE, TEARDOWN_ATTRIBUTE,
                             Project, NAME_ATTRIBUTE, ENVIRONMENTS_ATTRIBUTE, optional)
 from pybuilder.errors import PyBuilderException, ProjectValidationFailedException
-from pybuilder.execution import Action, Initializer, Task
+from pybuilder.execution import Action, Initializer, Task, TaskDependency
 from pybuilder.pluginloader import (BuiltinPluginLoader,
                                     DispatchingPluginLoader,
                                     ThirdPartyPluginLoader,
@@ -205,7 +205,7 @@ class Reactor(object):
         self.collect_tasks_and_actions_and_initializers(plugin_module)
 
     def collect_tasks_and_actions_and_initializers(self, project_module):
-        task_dependencies = dict()
+        injected_task_dependencies = dict()
 
         def normalize_candidate_name(candidate):
             if hasattr(candidate, NAME_ATTRIBUTE):
@@ -217,13 +217,12 @@ class Reactor(object):
             for name in as_list(names):
                 if not isinstance(name, basestring):
                     name = normalize_candidate_name(name)
-                if name not in task_dependencies:
-                    task_dependencies[name] = (list(), list())
-                task_dependencies[name][1 if optional else 0].append(depends_on)
+                if name not in injected_task_dependencies:
+                    injected_task_dependencies[name] = list()
+                injected_task_dependencies[name].append(TaskDependency(depends_on, optional))
 
         for name in dir(project_module):
             candidate = getattr(project_module, name)
-            name = normalize_candidate_name(candidate)
 
             if hasattr(candidate, TASK_ATTRIBUTE) and getattr(candidate, TASK_ATTRIBUTE):
                 dependents = getattr(candidate, DEPENDENTS_ATTRIBUTE) if hasattr(candidate,
@@ -231,7 +230,7 @@ class Reactor(object):
                 if dependents:
                     dependents = list(as_list(dependents))
                     for d in dependents:
-                        if type(d) is optional:
+                        if isinstance(d, optional):
                             d = d()
                             add_task_dependency(d, candidate, True)
                         else:
@@ -246,27 +245,24 @@ class Reactor(object):
 
             if hasattr(candidate, TASK_ATTRIBUTE) and getattr(candidate, TASK_ATTRIBUTE):
                 dependencies = getattr(candidate, DEPENDS_ATTRIBUTE) if hasattr(candidate, DEPENDS_ATTRIBUTE) else None
-                required_dependencies = []
-                optional_dependencies = []
-
-                # Add injected
-                if name in task_dependencies:
-                    injected_dependencies = task_dependencies[name]
-                    required_dependencies.extend(injected_dependencies[0])
-                    optional_dependencies.extend(injected_dependencies[1])
 
+                task_dependencies = list()
                 if dependencies:
                     dependencies = list(as_list(dependencies))
                     for d in dependencies:
-                        if type(d) is optional:
+                        if isinstance(d, optional):
                             d = as_list(d())
-                            optional_dependencies += d
+                            task_dependencies.extend([TaskDependency(item, True) for item in d])
                         else:
-                            required_dependencies.append(d)
-                self.logger.debug("Found task '%s' with required dependencies %s and optional dependencies %s", name,
-                                  required_dependencies, optional_dependencies)
+                            task_dependencies.append(TaskDependency(d))
+
+                # Add injected
+                if name in injected_task_dependencies:
+                    task_dependencies.extend(injected_task_dependencies[name])
+
+                self.logger.debug("Found task '%s' with dependencies %s", name, task_dependencies)
                 self.execution_manager.register_task(
-                    Task(name, candidate, required_dependencies, description, optional_dependencies))
+                    Task(name, candidate, task_dependencies, description))
 
             elif hasattr(candidate, ACTION_ATTRIBUTE) and getattr(candidate, ACTION_ATTRIBUTE):
                 before = getattr(candidate, BEFORE_ATTRIBUTE) if hasattr(
diff --git a/src/unittest/python/execution_tests.py b/src/unittest/python/execution_tests.py
index 08fd5b1..3775d05 100644
--- a/src/unittest/python/execution_tests.py
+++ b/src/unittest/python/execution_tests.py
@@ -20,12 +20,12 @@ import unittest
 
 from mockito import verify, unstub, any, times, when
 
-from test_utils import mock
+from pybuilder.core import Logger
 from pybuilder.errors import MissingTaskDependencyException, CircularTaskDependencyException, NoSuchTaskException, \\
     MissingActionDependencyException, InvalidNameException, RequiredTaskExclusionException
-from pybuilder.core import Logger
 from pybuilder.execution import as_task_name_list, Action, Executable, ExecutionManager, Task, \\
-    DependenciesNotResolvedException, Initializer
+    DependenciesNotResolvedException, Initializer, TaskDependency
+from test_utils import mock
 
 
 class AsTaskNameList(unittest.TestCase):
@@ -311,7 +311,7 @@ class ExecutionManagerTaskTest(ExecutionManagerTestBase):
                           mock())
 
     def test_ensure_task_is_executed_when_calling_execute_task(self):
-        task = mock(name="spam", dependencies=[], optional_dependencies=[])
+        task = mock(name="spam", dependencies=[])
 
         self.execution_manager.register_task(task)
         self.execution_manager.resolve_dependencies()
@@ -321,7 +321,7 @@ class ExecutionManagerTaskTest(ExecutionManagerTestBase):
         verify(task).execute(any(), {"a": 1})
 
     def test_ensure_before_action_is_executed_when_task_is_executed(self):
-        task = mock(name="task", dependencies=[], optional_dependencies=[])
+        task = mock(name="task", dependencies=[])
         action = mock(name="action", execute_before=["task"], execute_after=[])
 
         self.execution_manager.register_action(action)
@@ -334,7 +334,7 @@ class ExecutionManagerTaskTest(ExecutionManagerTestBase):
         verify(task).execute(any(), {})
 
     def test_ensure_after_action_is_executed_when_task_is_executed(self):
-        task = mock(name="task", dependencies=[], optional_dependencies=[])
+        task = mock(name="task", dependencies=[])
         action = mock(name="action", execute_before=[], execute_after=["task"])
 
         self.execution_manager.register_action(action)
@@ -347,7 +347,7 @@ class ExecutionManagerTaskTest(ExecutionManagerTestBase):
         verify(task).execute(any(), {})
 
     def test_ensure_after_action_teardown_is_executed_when_task_fails(self):
-        task = mock(name="task", dependencies=[], optional_dependencies=[])
+        task = mock(name="task", dependencies=[])
         when(task).execute(any(), {}).thenRaise(ValueError("simulated task error"))
         action = mock(name="action", execute_before=[], execute_after=["task"], teardown=True)
 
@@ -366,7 +366,7 @@ class ExecutionManagerTaskTest(ExecutionManagerTestBase):
         verify(task).execute(any(), {})
 
     def test_ensure_after_action_teardown_is_executed_when_action_fails(self):
-        task = mock(name="task", dependencies=[], optional_dependencies=[])
+        task = mock(name="task", dependencies=[])
         action_regular = mock(name="action_regular", execute_before=[], execute_after=["task"], teardown=False)
         when(action_regular).execute({}).thenRaise(ValueError("simulated action error"))
         action_teardown = mock(name="action_teardown", execute_before=[], execute_after=["task"], teardown=True)
@@ -392,7 +392,7 @@ class ExecutionManagerTaskTest(ExecutionManagerTestBase):
         verify(action_after_teardown, times(0)).execute({})
 
     def test_ensure_after_action_teardown_suppression_works_when_action_fails(self):
-        task = mock(name="task", dependencies=[], optional_dependencies=[])
+        task = mock(name="task", dependencies=[])
         action_regular = mock(name="action_regular", execute_before=[], execute_after=["task"], teardown=False)
         when(action_regular).execute({}).thenRaise(ValueError("simulated action error"))
         action_teardown = mock(name="action_teardown", execute_before=[], execute_after=["task"], teardown=True)
@@ -418,7 +418,7 @@ class ExecutionManagerTaskTest(ExecutionManagerTestBase):
         verify(action_after_teardown, times(0)).execute({})
 
     def test_ensure_after_action_teardown_is_executed_and_suppresses(self):
-        task = mock(name="task", dependencies=[], optional_dependencies=[])
+        task = mock(name="task", dependencies=[])
         when(task).execute(any(), {}).thenRaise(ValueError("simulated task error"))
         action_teardown1 = mock(name="action_teardown1", execute_before=[], execute_after=["task"], teardown=True,
                                 source="task")
@@ -462,8 +462,8 @@ class ExecutionManagerActionTest(ExecutionManagerTestBase):
         self.assertEquals({"action": action}, self.execution_manager._actions)
 
     def test_ensure_action_registered_for_two_tasks_is_executed_two_times(self):
-        spam = mock(name="spam", dependencies=[], optional_dependencies=[])
-        eggs = mock(name="eggs", dependencies=[], optional_dependencies=[])
+        spam = mock(name="spam", dependencies=[])
+        eggs = mock(name="eggs", dependencies=[])
         self.execution_manager.register_task(spam, eggs)
 
         action = mock(name="action",
@@ -479,8 +479,8 @@ class ExecutionManagerActionTest(ExecutionManagerTestBase):
         verify(action, times(2)).execute(any())
 
     def test_ensure_action_registered_for_two_tasks_is_executed_only_once_if_single_attribute_is_present(self):
-        spam = mock(name="spam", dependencies=[], optional_dependencies=[])
-        eggs = mock(name="eggs", dependencies=[], optional_dependencies=[])
+        spam = mock(name="spam", dependencies=[])
+        eggs = mock(name="eggs", dependencies=[])
         self.execution_manager.register_task(spam, eggs)
 
         action = mock(name="action",
@@ -502,7 +502,7 @@ class ExecutionManagerResolveDependenciesTest(ExecutionManagerTestBase):
         self.assertTrue(self.execution_manager._dependencies_resolved)
 
     def test_ensure_that_dependencies_are_resolved_when_single_task_is_given(self):
-        task = mock(dependencies=[], optional_dependencies=[])
+        task = mock(dependencies=[])
 
         self.execution_manager.register_task(task)
 
@@ -510,7 +510,7 @@ class ExecutionManagerResolveDependenciesTest(ExecutionManagerTestBase):
         self.assertTrue(self.execution_manager._dependencies_resolved)
 
     def test_should_raise_exception_when_task_depends_on_task_not_found(self):
-        task = mock(dependencies=["not_found"])
+        task = mock(dependencies=[TaskDependency("not_found")])
 
         self.execution_manager.register_task(task)
 
@@ -534,105 +534,109 @@ class ExecutionManagerResolveDependenciesTest(ExecutionManagerTestBase):
                           self.execution_manager.resolve_dependencies)
 
     def test_ensure_that_dependencies_are_resolved_when_simple_dependency_is_found(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=["one"], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
 
         self.execution_manager.register_task(one, two)
-
         self.execution_manager.resolve_dependencies()
 
-        self.assertEquals(
-            [], self.execution_manager._task_dependencies.get("one"))
-        self.assertEquals(
-            [one], self.execution_manager._task_dependencies.get("two"))
+        self.assertEquals([], self.execution_manager._task_dependencies.get("one"))
+        self.assertEquals([TaskDependency(one)], self.execution_manager._task_dependencies.get("two"))
 
     def test_ensure_that_dependencies_are_resolved_when_task_depends_on_multiple_tasks(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=["one"], optional_dependencies=[])
-        three = mock(name="three", dependencies=["one", "two"], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
+        three = mock(name="three", dependencies=[TaskDependency("one"), TaskDependency("two")])
 
         self.execution_manager.register_task(one, two, three)
-
         self.execution_manager.resolve_dependencies()
 
+        self.assertEquals([], self.execution_manager._task_dependencies.get("one"))
+        self.assertEquals([TaskDependency(one)], self.execution_manager._task_dependencies.get("two"))
         self.assertEquals(
-            [], self.execution_manager._task_dependencies.get("one"))
-        self.assertEquals(
-            [one], self.execution_manager._task_dependencies.get("two"))
-        self.assertEquals(
-            [one, two], self.execution_manager._task_dependencies.get("three"))
+            [TaskDependency(one), TaskDependency(two)], self.execution_manager._task_dependencies.get("three"))
 
-    def test_ensure_that_optional_tasks_are_excluded(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=[], optional_dependencies=["one"])
+    def test_override_optional_dependency_with_required(self):
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one", True), TaskDependency("one")])
 
         self.execution_manager.register_task(one, two)
+        self.execution_manager.resolve_dependencies()
 
-        self.execution_manager.resolve_dependencies(exclude_optional_tasks=["one"])
-
-        self.assertEquals(
-            [], self.execution_manager._task_dependencies.get("one"))
-        self.assertEquals(
-            [], self.execution_manager._task_dependencies.get("two"))
+        self.assertEquals([], self.execution_manager._task_dependencies.get("one"))
+        self.assertEquals([TaskDependency(one)], self.execution_manager._task_dependencies.get("two"))
 
-    def test_ensure_that_required_tasks_are_force_excluded(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=["one"], optional_dependencies=[])
+    def test_ignore_required_override_with_optional_dependency(self):
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one"), TaskDependency("one", True)])
 
         self.execution_manager.register_task(one, two)
+        self.execution_manager.resolve_dependencies()
 
-        self.execution_manager.resolve_dependencies(exclude_tasks=["one"])
+        self.assertEquals([], self.execution_manager._task_dependencies.get("one"))
+        self.assertEquals([TaskDependency(one)], self.execution_manager._task_dependencies.get("two"))
 
-        self.assertEquals(
-            [], self.execution_manager._task_dependencies.get("one"))
-        self.assertEquals(
-            [], self.execution_manager._task_dependencies.get("two"))
+    def test_ignore_second_optional_dependency(self):
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one", True), TaskDependency("one", True)])
 
-    def test_ensure_that_required_tasks_are_not_optionally_excluded(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=["one"], optional_dependencies=[])
+        self.execution_manager.register_task(one, two)
+        self.execution_manager.resolve_dependencies()
+
+        self.assertEquals([], self.execution_manager._task_dependencies.get("one"))
+        self.assertEquals([TaskDependency(one, True)], self.execution_manager._task_dependencies.get("two"))
+
+    def test_ignore_second_required_dependency(self):
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one"), TaskDependency("one")])
 
         self.execution_manager.register_task(one, two)
+        self.execution_manager.resolve_dependencies()
 
-        self.assertRaises(RequiredTaskExclusionException, self.execution_manager.resolve_dependencies,
-                          exclude_optional_tasks=["one"])
+        self.assertEquals([], self.execution_manager._task_dependencies.get("one"))
+        self.assertEquals([TaskDependency(one)], self.execution_manager._task_dependencies.get("two"))
 
 
 class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
     def test_should_collect_all_tasks_when_there_are_no_dependencies(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
         self.execution_manager.register_task(one)
+        self.execution_manager.resolve_dependencies()
 
         self.assertEqual(self.execution_manager.collect_all_transitive_tasks(["one"]), set([one]))
 
     def test_should_collect_all_tasks_when_there_is_a_simple_dependency(self):
-        one = mock(name="one", dependencies=["two"])
-        two = mock(name="two", dependencies=[], optional_dependencies=[])
+        one = mock(name="one", dependencies=[TaskDependency("two")])
+        two = mock(name="two", dependencies=[])
         self.execution_manager.register_task(one, two)
+        self.execution_manager.resolve_dependencies()
 
         self.assertEqual(self.execution_manager.collect_all_transitive_tasks(["one"]), set([one, two]))
 
     def test_should_collect_all_tasks_when_there_is_a_transitive_dependency(self):
-        one = mock(name="one", dependencies=["two"])
-        two = mock(name="two", dependencies=["three"])
-        three = mock(name="three", dependencies=[], optional_dependencies=[])
+        one = mock(name="one", dependencies=[TaskDependency("two")])
+        two = mock(name="two", dependencies=[TaskDependency("three")])
+        three = mock(name="three", dependencies=[])
         self.execution_manager.register_task(one, two, three)
+        self.execution_manager.resolve_dependencies()
 
         self.assertEqual(self.execution_manager.collect_all_transitive_tasks(["one"]), set([one, two, three]))
 
     def test_should_collect_all_tasks_when_several_tasks_given(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=["three"])
-        three = mock(name="three", dependencies=[], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("three")])
+        three = mock(name="three", dependencies=[])
         self.execution_manager.register_task(one, two, three)
+        self.execution_manager.resolve_dependencies()
 
         self.assertEqual(self.execution_manager.collect_all_transitive_tasks(["one", "two"]), set([one, two, three]))
 
     def test_should_only_collect_required_tasks(self):
-        one = mock(name="one", dependencies=["three"])
-        two = mock(name="two", dependencies=[], optional_dependencies=[])
-        three = mock(name="three", dependencies=[], optional_dependencies=[])
+        one = mock(name="one", dependencies=[TaskDependency("three")])
+        two = mock(name="two", dependencies=[])
+        three = mock(name="three", dependencies=[])
         self.execution_manager.register_task(one, two, three)
+        self.execution_manager.resolve_dependencies()
 
         self.assertEqual(self.execution_manager.collect_all_transitive_tasks(["one"]), set([one, three]))
 
@@ -646,7 +650,7 @@ class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
             NoSuchTaskException, self.execution_manager.build_execution_plan, ("boom",))
 
     def test_should_return_execution_plan_with_single_task_when_single_task_is_to_be_executed(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
 
         self.execution_manager.register_task(one)
         self.execution_manager.resolve_dependencies()
@@ -655,8 +659,8 @@ class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
             [one], self.execution_manager.build_execution_plan(["one"]))
 
     def test_should_return_execution_plan_with_two_tasks_when_two_tasks_are_to_be_executed(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=[], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[])
 
         self.execution_manager.register_task(one, two)
         self.execution_manager.resolve_dependencies()
@@ -665,8 +669,8 @@ class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
             [one, two], self.execution_manager.build_execution_plan(["one", "two"]))
 
     def test_ensure_that_dependencies_are_executed_before_root_task(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=["one"], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
 
         self.execution_manager.register_task(one, two)
         self.execution_manager.resolve_dependencies()
@@ -675,7 +679,7 @@ class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
             [one, two], self.execution_manager.build_execution_plan(["two"]))
 
     def test_ensure_that_tasks_are_not_executed_multiple_times(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
+        one = mock(name="one", dependencies=[], )
 
         self.execution_manager.register_task(one)
         self.execution_manager.resolve_dependencies()
@@ -684,8 +688,8 @@ class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
             [one], self.execution_manager.build_execution_plan(["one", "one"]))
 
     def test_ensure_that_tasks_are_not_executed_multiple_times_when_being_dependencies(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=["one"], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
 
         self.execution_manager.register_task(one, two)
         self.execution_manager.resolve_dependencies()
@@ -694,7 +698,7 @@ class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
             [one, two], self.execution_manager.build_execution_plan(["one", "two"]))
 
     def test_should_raise_exception_when_circular_reference_is_detected_on_single_task(self):
-        one = mock(name="one", dependencies=["one"], optional_dependencies=[])
+        one = mock(name="one", dependencies=[TaskDependency("one")])
 
         self.execution_manager.register_task(one)
         self.execution_manager.resolve_dependencies()
@@ -703,20 +707,19 @@ class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
                           self.execution_manager.build_execution_plan, ["one"])
 
     def test_should_raise_exception_when_circular_reference_is_detected_on_two_tasks(self):
-        one = mock(name="one", dependencies=["two"], optional_dependencies=[])
-        two = mock(name="two", dependencies=["one"], optional_dependencies=[])
+        one = mock(name="one", dependencies=[TaskDependency("two")])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
 
         self.execution_manager.register_task(one, two)
-
         self.execution_manager.resolve_dependencies()
 
         self.assertRaises(CircularTaskDependencyException,
                           self.execution_manager.build_execution_plan, ["one"])
 
     def test_should_raise_exception_when_circular_reference_is_detected_on_three_tasks(self):
-        one = mock(name="one", dependencies=["three"], optional_dependencies=[])
-        two = mock(name="two", dependencies=["one"], optional_dependencies=[])
-        three = mock(name="three", dependencies=["one", "two"], optional_dependencies=[])
+        one = mock(name="one", dependencies=[TaskDependency("three")])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
+        three = mock(name="three", dependencies=[TaskDependency("one"), TaskDependency("two")])
 
         self.execution_manager.register_task(one, two, three)
 
@@ -726,9 +729,9 @@ class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
                           self.execution_manager.build_execution_plan, ["one"])
 
     def test_should_raise_exception_when_circular_reference_is_detected_on_indirect_required_tasks(self):
-        one = mock(name="one", dependencies=["two"], optional_dependencies=[])
-        two = mock(name="two", dependencies=["three"], optional_dependencies=[])
-        three = mock(name="three", dependencies=["two"], optional_dependencies=[])
+        one = mock(name="one", dependencies=[TaskDependency("two")])
+        two = mock(name="two", dependencies=[TaskDependency("three")])
+        three = mock(name="three", dependencies=[TaskDependency("two")])
 
         self.execution_manager.register_task(one, two, three)
 
@@ -738,9 +741,9 @@ class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
                           self.execution_manager.build_execution_plan, ["one"])
 
     def test_shortest_execution_plan_is_shortest(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=["one"], optional_dependencies=[])
-        three = mock(name="three", dependencies=["two"], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
+        three = mock(name="three", dependencies=[TaskDependency("two")])
 
         self.execution_manager.register_task(one, two, three)
         self.execution_manager.resolve_dependencies()
@@ -753,9 +756,9 @@ class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
         self.assertEquals([three], self.execution_manager.build_shortest_execution_plan("three"))
 
     def test_shortest_execution_plan_always_executes_target(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=["one"], optional_dependencies=[])
-        three = mock(name="three", dependencies=["two"], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
+        three = mock(name="three", dependencies=[TaskDependency("two")])
 
         self.execution_manager.register_task(one, two, three)
         self.execution_manager.resolve_dependencies()
@@ -769,9 +772,9 @@ class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
         self.assertEquals([three], self.execution_manager.build_shortest_execution_plan("three"))
 
     def test_shortest_execution_plan_checks_circularity(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=["one"], optional_dependencies=[])
-        three = mock(name="three", dependencies=["two"], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
+        three = mock(name="three", dependencies=[TaskDependency("two")])
 
         self.execution_manager.register_task(one, two, three)
         self.execution_manager.resolve_dependencies()
@@ -783,9 +786,9 @@ class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
                           ["three"])
 
     def test_shortest_execution_plan_reruns_on_demand(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=["one"], optional_dependencies=[])
-        three = mock(name="three", dependencies=["two"], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
+        three = mock(name="three", dependencies=[TaskDependency("two")])
 
         self.execution_manager.register_task(one, two, three)
         self.execution_manager.resolve_dependencies()
@@ -797,6 +800,71 @@ class ExecutionManagerBuildExecutionPlanTest(ExecutionManagerTestBase):
         self.assertEquals([two, three], self.execution_manager.build_shortest_execution_plan(("three", "two")))
         self.assertEquals([one, two, three], self.execution_manager.build_shortest_execution_plan(("three", "one")))
 
+    def test_ensure_that_optional_tasks_are_excluded(self):
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one", True)])
+
+        self.execution_manager.register_task(one, two)
+        self.execution_manager.resolve_dependencies(exclude_optional_tasks=["one"])
+
+        execution_plan = self.execution_manager.build_execution_plan("two")
+
+        self.assertEquals([two], execution_plan)
+
+    def test_ensure_that_optional_branch_is_excluded(self):
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
+        three = mock(name="three", dependencies=[TaskDependency("two", True)])
+
+        self.execution_manager.register_task(one, two, three)
+        self.execution_manager.resolve_dependencies(exclude_optional_tasks=["two"])
+
+        execution_plan = self.execution_manager.build_execution_plan("three")
+
+        self.assertEquals([three], execution_plan)
+
+    def test_ensure_that_required_tasks_are_force_excluded(self):
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
+
+        self.execution_manager.register_task(one, two)
+        self.execution_manager.resolve_dependencies(exclude_tasks=["one"])
+
+        execution_plan = self.execution_manager.build_execution_plan("two")
+
+        self.assertEquals([two], execution_plan)
+
+    def test_ensure_that_required_branch_is_force_excluded(self):
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
+        three = mock(name="three", dependencies=[TaskDependency("two")])
+
+        self.execution_manager.register_task(one, two, three)
+        self.execution_manager.resolve_dependencies(exclude_tasks=["two"])
+
+        execution_plan = self.execution_manager.build_execution_plan("three")
+
+        self.assertEquals([three], execution_plan)
+
+    def test_ensure_that_required_tasks_are_not_optionally_excluded(self):
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
+
+        self.execution_manager.register_task(one, two)
+        self.execution_manager.resolve_dependencies(exclude_optional_tasks=["one"])
+
+        self.assertRaises(RequiredTaskExclusionException, self.execution_manager.build_execution_plan, "two")
+
+    def test_ensure_that_required_tasks_branch_not_optionally_excluded(self):
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
+        three = mock(name="three", dependencies=[TaskDependency("two")])
+
+        self.execution_manager.register_task(one, two, three)
+        self.execution_manager.resolve_dependencies(exclude_optional_tasks=["one"])
+
+        self.assertRaises(RequiredTaskExclusionException, self.execution_manager.build_execution_plan, "three")
+
 
 class ExecutionManagerExecuteExecutionPlanTest(ExecutionManagerTestBase):
     def test_should_raise_exception_when_dependencies_are_not_resolved(self):
@@ -804,8 +872,8 @@ class ExecutionManagerExecuteExecutionPlanTest(ExecutionManagerTestBase):
                           self.execution_manager.execute_execution_plan, ["boom"])
 
     def test_ensure_tasks_are_executed(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=[], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[])
 
         self.execution_manager.register_task(one, two)
         self.execution_manager.resolve_dependencies()
@@ -816,9 +884,9 @@ class ExecutionManagerExecuteExecutionPlanTest(ExecutionManagerTestBase):
         verify(two).execute(any(), {})
 
     def test_shortest_execution_plan_executed(self):
-        one = mock(name="one", dependencies=[], optional_dependencies=[])
-        two = mock(name="two", dependencies=["one"], optional_dependencies=[])
-        three = mock(name="three", dependencies=["two"], optional_dependencies=[])
+        one = mock(name="one", dependencies=[])
+        two = mock(name="two", dependencies=[TaskDependency("one")])
+        three = mock(name="three", dependencies=[TaskDependency("two")])
 
         self.execution_manager.register_task(one, two, three)
         self.execution_manager.resolve_dependencies()
diff --git a/src/unittest/python/reactor_tests.py b/src/unittest/python/reactor_tests.py
index f18625b..c969baa 100644
--- a/src/unittest/python/reactor_tests.py
+++ b/src/unittest/python/reactor_tests.py
@@ -35,7 +35,7 @@ from pybuilder.core import (ENVIRONMENTS_ATTRIBUTE,
                             after,
                             before)
 from pybuilder.errors import MissingPluginException, PyBuilderException, ProjectValidationFailedException
-from pybuilder.execution import Task, Action, Initializer, ExecutionManager
+from pybuilder.execution import Task, TaskDependency, Action, Initializer, ExecutionManager
 from pybuilder.pluginloader import PluginLoader
 from pybuilder.reactor import Reactor
 from test_utils import mock  # TODO @mriehl SORCERY!!!!! BURN IT WITH FIRE!!!!
@@ -169,12 +169,13 @@ class ReactorTest(unittest.TestCase):
 
         self.reactor.collect_tasks_and_actions_and_initializers(module)
 
-        inorder.verify(pybuilder.reactor).Task("task1", task1, [], '', [])
-        inorder.verify(pybuilder.reactor).Task("task2", task2, [task1], '', [])
-        inorder.verify(pybuilder.reactor).Task("task3", task3, [], '', [task5])
-        inorder.verify(pybuilder.reactor).Task("task4", task4, [], '', [task3])
-        inorder.verify(pybuilder.reactor).Task("task5", task5, [], '', [])
-        inorder.verify(pybuilder.reactor).Task("task6", task6, [task4, task5, task1], '', [task2])
+        inorder.verify(pybuilder.reactor).Task("task1", task1, [], '')
+        inorder.verify(pybuilder.reactor).Task("task2", task2, [TaskDependency(task1)], '')
+        inorder.verify(pybuilder.reactor).Task("task3", task3, [TaskDependency(task5, True)], '')
+        inorder.verify(pybuilder.reactor).Task("task4", task4, [TaskDependency(task3, True)], '')
+        inorder.verify(pybuilder.reactor).Task("task5", task5, [], '')
+        inorder.verify(pybuilder.reactor).Task("task6", task6, [TaskDependency(task1), TaskDependency(task2, True),
+                                                                TaskDependency(task4), TaskDependency(task5)], '')
 
     def test_should_collect_single_before_action(self):
         @before("spam")
['setup.py', 'src/integrationtest/python/should_raise_exception_optional_exclude_required_task_tests.py', 'src/unittest/python/reactor_tests.py', 'src/main/python/pybuilder/cli.py', 'src/main/python/pybuilder/reactor.py', 'src/unittest/python/execution_tests.py', 'src/main/python/pybuilder/execution.py']