Repository: PrefectHQ/prefect

Issue:
FlowRunner won't preserve all items from prefect.context
## Description
<!-- A clear description of the bug -->
FlowRunner does not preserve all items from prefect.context (e.g. dates: today, yesterday, etc.), because `initialize_run()` method prepares an empty context if not specified in `context` parameter. This could be okay, but contradicts to the [documentation](https://docs.prefect.io/core/concepts/execution.html#modifying-context-at-runtime), where it is suggested that for "backfills", dates could be overridded (e.g. today). 

## Expected Behavior
<!-- What did you expect to happen instead? -->
Items already set in prefect.context should be respected.

## Reproduction
<!-- A minimal example that exhibits the behavior. -->

Both runs should output the same value "YESTERDAAAAAY" for `yesterday`, but the first outputs the actual date (2021-03-18 as of this writing).

```python
import prefect
from prefect import task, Flow

CTX = dict(test_variable="TEST_VALUE", yesterday="YESTERDAAAAAY")

@task(log_stdout=True)
def print_env():
    print('TEST_VARIABLE: %s' % prefect.context.get('test_variable', 'MISSING'))
    print('YESTERDAY: %s' % prefect.context.get('yesterday', 'MISSING'))

with Flow("Context override test") as flow:
    print_env()

## Run 1
with prefect.context(**CTX):
    flow.run()

## Run 2
flow.run(context=CTX)
```

## Environment
<!-- Any additional information about your environment

Optionally run `prefect diagnostics` from the command line and paste the information here. -->
```json
{
  "config_overrides": {},
  "env_vars": [],
  "system_information": {
    "platform": "Linux-4.18.0-240.10.1.el8_3.x86_64-x86_64-with-centos-8.3.2011",
    "prefect_backend": "server",
    "prefect_version": "0.14.10",
    "python_version": "3.6.8"
  }
}
```

## Suggested Resolution

A single line added to FlowRunner could solve that. 

```diff
--- prefect/engine/flow_runner.py.orig  2021-02-16 16:08:28.204208267 +0100
+++ prefect/engine/flow_runner.py       2021-03-19 16:36:56.278923592 +0100
@@ -152,6 +152,7 @@
         for param, value in (parameters or {}).items():
             context_params[param] = value
 
+        context.update(prefect.context)
         context.update(flow_name=self.flow.name)
         context.setdefault("scheduled_start_time", pendulum.now("utc"))
```

Discussion:

Patch:
diff --git a/changes/pr4287.yaml b/changes/pr4287.yaml
new file mode 100644
index 0000000000..2f28d5641f
--- /dev/null
+++ b/changes/pr4287.yaml
@@ -0,0 +1,2 @@
+fix:
+  - "Fix bug where sometimes the global `prefect.context` wouldn't be respected during a flow run - [#4287](https://github.com/PrefectHQ/prefect/pull/4287)"
diff --git a/src/prefect/engine/flow_runner.py b/src/prefect/engine/flow_runner.py
index 5150b15fb2..e741366341 100644
--- a/src/prefect/engine/flow_runner.py
+++ b/src/prefect/engine/flow_runner.py
@@ -224,10 +224,13 @@ class FlowRunner(Runner):
         self.logger.info("Beginning Flow run for '{}'".format(self.flow.name))
 
         # make copies to avoid modifying user inputs
+        parameters = dict(parameters or {})
         task_states = dict(task_states or {})
-        context = dict(context or {})
         task_contexts = dict(task_contexts or {})
-        parameters = dict(parameters or {})
+        # Default to global context, with provided context as override
+        run_context = dict(prefect.context)
+        run_context.update(context or {})
+
         if executor is None:
             # Use the executor on the flow, if configured
             executor = getattr(self.flow, "executor", None)
@@ -237,15 +240,15 @@ class FlowRunner(Runner):
         self.logger.debug("Using executor type %s", type(executor).__name__)
 
         try:
-            state, task_states, context, task_contexts = self.initialize_run(
+            state, task_states, run_context, task_contexts = self.initialize_run(
                 state=state,
                 task_states=task_states,
-                context=context,
+                context=run_context,
                 task_contexts=task_contexts,
                 parameters=parameters,
             )
 
-            with prefect.context(context):
+            with prefect.context(run_context):
                 state = self.check_flow_is_pending_or_running(state)
                 state = self.check_flow_reached_start_time(state)
                 state = self.set_flow_to_running(state)
@@ -266,7 +269,7 @@ class FlowRunner(Runner):
             self.logger.exception(
                 "Unexpected error while running flow: {}".format(repr(exc))
             )
-            if prefect.context.get("raise_on_exception"):
+            if run_context.get("raise_on_exception"):
                 raise exc
             new_state = Failed(
                 message="Unexpected error while running flow: {}".format(repr(exc)),


Test patch:
diff --git a/tests/engine/test_flow_runner.py b/tests/engine/test_flow_runner.py
index 47749cd250..c07cc95fbf 100644
--- a/tests/engine/test_flow_runner.py
+++ b/tests/engine/test_flow_runner.py
@@ -1323,18 +1323,28 @@ class TestContext:
         output = res.result[return_ctx_key].result
         assert isinstance(output, datetime.datetime)
 
-    def test_user_provided_context_is_prioritized(self):
+    @pytest.mark.parametrize(
+        "outer_context, inner_context, sol",
+        [
+            ({"date": "outer"}, {"date": "inner"}, "inner"),
+            ({"date": "outer"}, {}, "outer"),
+        ],
+    )
+    def test_user_provided_context_is_prioritized(
+        self, outer_context, inner_context, sol
+    ):
         @prefect.task
         def return_ctx_key():
             return prefect.context.get("date")
 
         f = Flow(name="test", tasks=[return_ctx_key])
-        res = f.run(context={"date": "42"})
+        with prefect.context(**outer_context):
+            res = f.run(context=inner_context)
 
         assert res.is_successful()
 
         output = res.result[return_ctx_key].result
-        assert output == "42"
+        assert output == sol
 
 
 @pytest.mark.parametrize(