prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Add audio to the ingestion server tests
  Audio is currently not included in the ingestion server integration or unit tests. We should update these tests to include support for audio. Separate PRs for unit and integration tests would be best. Below is some information on these tests and how to work with them.
  
  ## Running the tests 
  
  To run the tests and get a sense of what they do, do the following steps:
  
  ```bash
  cd ingestion_server
  pipenv install
  pipenv run python3 test/integration_tests.py
  ```
  
  This is currently blocked by #143. I would've liked to run the tests to learn a bit more about how they work but this isn't yet possible.
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `sample_data/make_sample_pop.py`
  Content:
  ```
  1 import csv
  2 import random
  3 
  4 
  5 in_tsv = open("sample_data.csv", "r")
  6 out_tsv = open("sample_popularity_data.csv", "w+")
  7 output_fields = ["identifier", "normalized_popularity"]
  8 reader = csv.DictReader(in_tsv, delimiter=",")
  9 writer = csv.DictWriter(out_tsv, delimiter=",", fieldnames=output_fields)
  10 writer.writeheader()
  11 for row in reader:
  12     pop = random.uniform(0, 100)
  13     out_row = {"identifier": row["identifier"], "normalized_popularity": pop}
  14     writer.writerow(out_row)
  15
  ```
  
  Path: `ingestion_server/ingestion_server/tasks.py`
  Content:
  ```
  1 """
  2 Simple in-memory tracking of executed tasks.
  3 """
  4 
  5 import datetime as dt
  6 import logging
  7 from enum import Enum
  8 from multiprocessing import Process
  9 
  10 import requests
  11 
  12 from ingestion_server.indexer import TableIndexer, elasticsearch_connect
  13 from ingestion_server.ingest import reload_upstream
  14 
  15 
  16 class TaskTypes(Enum):
  17     # Completely reindex all data for a given model.
  18     REINDEX = 0
  19     # Reindex updates to a model from the database since a certain date.
  20     UPDATE_INDEX = 1
  21     # Download the latest copy of the data from the upstream database, then
  22     # completely reindex the newly imported data.
  23     INGEST_UPSTREAM = 2
  24     # Create indices in Elasticsearch for QA tests.
  25     # This is not intended for production use, but can be safely executed in a
  26     # production environment without consequence.
  27     LOAD_TEST_DATA = 3
  28 
  29 
  30 class TaskTracker:
  31     def __init__(self):
  32         self.id_task = {}
  33         self.id_action = {}
  34         self.id_progress = {}
  35         self.id_start_time = {}
  36         self.id_finish_time = {}
  37 
  38     def add_task(self, task, task_id, action, progress, finish_time):
  39         self._prune_old_tasks()
  40         self.id_task[task_id] = task
  41         self.id_action[task_id] = action
  42         self.id_progress[task_id] = progress
  43         self.id_start_time[task_id] = dt.datetime.utcnow().timestamp()
  44         self.id_finish_time[task_id] = finish_time
  45         return task_id
  46 
  47     def _prune_old_tasks(self):
  48         pass
  49 
  50     def list_task_statuses(self):
  51         self._prune_old_tasks()
  52         results = []
  53         for _id, task in self.id_task.items():
  54             percent_completed = self.id_progress[_id].value
  55             active = task.is_alive()
  56             start_time = self.id_start_time[_id]
  57             finish_time = self.id_finish_time[_id].value
  58             results.append(
  59                 {
  60                     "task_id": _id,
  61                     "active": active,
  62                     "action": self.id_action[_id],
  63                     "progress": percent_completed,
  64                     "error": percent_completed < 100 and not active,
  65                     "start_time": start_time,
  66                     "finish_time": finish_time,
  67                 }
  68             )
  69         sorted_results = sorted(results, key=lambda x: x["finish_time"])
  70 
  71         to_utc = dt.datetime.utcfromtimestamp
  72 
  73         def render_date(x):
  74             return to_utc(x) if x != 0.0 else None
  75 
  76         # Convert date to a readable format
  77         for idx, task in enumerate(sorted_results):
  78             start_time = task["start_time"]
  79             finish_time = task["finish_time"]
  80             sorted_results[idx]["start_time"] = str(render_date(start_time))
  81             sorted_results[idx]["finish_time"] = str(render_date(finish_time))
  82 
  83         return sorted_results
  84 
  85 
  86 class Task(Process):
  87     def __init__(
  88         self, model, task_type, since_date, progress, task_id, finish_time, callback_url
  89     ):
  90         Process.__init__(self)
  91         self.model = model
  92         self.task_type = task_type
  93         self.since_date = since_date
  94         self.progress = progress
  95         self.task_id = task_id
  96         self.finish_time = finish_time
  97         self.callback_url = callback_url
  98 
  99     def run(self):
  100         # Map task types to actions.
  101         elasticsearch = elasticsearch_connect()
  102         indexer = TableIndexer(
  103             elasticsearch, self.model, self.progress, self.finish_time
  104         )
  105         if self.task_type == TaskTypes.REINDEX:
  106             indexer.reindex(self.model)
  107         elif self.task_type == TaskTypes.UPDATE_INDEX:
  108             indexer.update(self.model, self.since_date)
  109         elif self.task_type == TaskTypes.INGEST_UPSTREAM:
  110             reload_upstream(self.model)
  111             if self.model == "audio":
  112                 reload_upstream("audioset", approach="basic")
  113             indexer.reindex(self.model)
  114         elif self.task_type == TaskTypes.LOAD_TEST_DATA:
  115             indexer.load_test_data(self.model)
  116         logging.info(f"Task {self.task_id} exited.")
  117         if self.callback_url:
  118             try:
  119                 requests.post(self.callback_url)
  120             except requests.exceptions.RequestException as e:
  121                 logging.error("Failed to send callback!")
  122                 logging.error(e)
  123
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/ingestion_server/ingestion_server/tasks.py b/ingestion_server/ingestion_server/tasks.py
  --- a/ingestion_server/ingestion_server/tasks.py
  +++ b/ingestion_server/ingestion_server/tasks.py
  @@ -116,7 +116,9 @@
           logging.info(f"Task {self.task_id} exited.")
           if self.callback_url:
               try:
  -                requests.post(self.callback_url)
  +                logging.info("Sending callback request")
  +                res = requests.post(self.callback_url)
  +                logging.info(f"Response: {res.text}")
               except requests.exceptions.RequestException as e:
                   logging.error("Failed to send callback!")
                   logging.error(e)
  diff --git a/sample_data/make_sample_pop.py b/sample_data/make_sample_pop.py
  deleted file mode 100644
  --- a/sample_data/make_sample_pop.py
  +++ /dev/null
  @@ -1,14 +0,0 @@
  -import csv
  -import random
  -
  -
  -in_tsv = open("sample_data.csv", "r")
  -out_tsv = open("sample_popularity_data.csv", "w+")
  -output_fields = ["identifier", "normalized_popularity"]
  -reader = csv.DictReader(in_tsv, delimiter=",")
  -writer = csv.DictWriter(out_tsv, delimiter=",", fieldnames=output_fields)
  -writer.writeheader()
  -for row in reader:
  -    pop = random.uniform(0, 100)
  -    out_row = {"identifier": row["identifier"], "normalized_popularity": pop}
  -    writer.writerow(out_row)
  
verification_info: {"golden_diff": "diff --git a/ingestion_server/ingestion_server/tasks.py b/ingestion_server/ingestion_server/tasks.py\n--- a/ingestion_server/ingestion_server/tasks.py\n+++ b/ingestion_server/ingestion_server/tasks.py\n@@ -116,7 +116,9 @@\n         logging.info(f\"Task {self.task_id} exited.\")\n         if self.callback_url:\n             try:\n-                requests.post(self.callback_url)\n+                logging.info(\"Sending callback request\")\n+                res = requests.post(self.callback_url)\n+                logging.info(f\"Response: {res.text}\")\n             except requests.exceptions.RequestException as e:\n                 logging.error(\"Failed to send callback!\")\n                 logging.error(e)\ndiff --git a/sample_data/make_sample_pop.py b/sample_data/make_sample_pop.py\ndeleted file mode 100644\n--- a/sample_data/make_sample_pop.py\n+++ /dev/null\n@@ -1,14 +0,0 @@\n-import csv\n-import random\n-\n-\n-in_tsv = open(\"sample_data.csv\", \"r\")\n-out_tsv = open(\"sample_popularity_data.csv\", \"w+\")\n-output_fields = [\"identifier\", \"normalized_popularity\"]\n-reader = csv.DictReader(in_tsv, delimiter=\",\")\n-writer = csv.DictWriter(out_tsv, delimiter=\",\", fieldnames=output_fields)\n-writer.writeheader()\n-for row in reader:\n-    pop = random.uniform(0, 100)\n-    out_row = {\"identifier\": row[\"identifier\"], \"normalized_popularity\": pop}\n-    writer.writerow(out_row)\n", "issue": "Add audio to the ingestion server tests\nAudio is currently not included in the ingestion server integration or unit tests. We should update these tests to include support for audio. Separate PRs for unit and integration tests would be best. Below is some information on these tests and how to work with them.\r\n\r\n## Running the tests \r\n\r\nTo run the tests and get a sense of what they do, do the following steps:\r\n\r\n```bash\r\ncd ingestion_server\r\npipenv install\r\npipenv run python3 test/integration_tests.py\r\n```\r\n\r\nThis is currently blocked by #143. I would've liked to run the tests to learn a bit more about how they work but this isn't yet possible.\n", "before_files": [{"content": "import csv\nimport random\n\n\nin_tsv = open(\"sample_data.csv\", \"r\")\nout_tsv = open(\"sample_popularity_data.csv\", \"w+\")\noutput_fields = [\"identifier\", \"normalized_popularity\"]\nreader = csv.DictReader(in_tsv, delimiter=\",\")\nwriter = csv.DictWriter(out_tsv, delimiter=\",\", fieldnames=output_fields)\nwriter.writeheader()\nfor row in reader:\n    pop = random.uniform(0, 100)\n    out_row = {\"identifier\": row[\"identifier\"], \"normalized_popularity\": pop}\n    writer.writerow(out_row)\n", "path": "sample_data/make_sample_pop.py"}, {"content": "\"\"\"\nSimple in-memory tracking of executed tasks.\n\"\"\"\n\nimport datetime as dt\nimport logging\nfrom enum import Enum\nfrom multiprocessing import Process\n\nimport requests\n\nfrom ingestion_server.indexer import TableIndexer, elasticsearch_connect\nfrom ingestion_server.ingest import reload_upstream\n\n\nclass TaskTypes(Enum):\n    # Completely reindex all data for a given model.\n    REINDEX = 0\n    # Reindex updates to a model from the database since a certain date.\n    UPDATE_INDEX = 1\n    # Download the latest copy of the data from the upstream database, then\n    # completely reindex the newly imported data.\n    INGEST_UPSTREAM = 2\n    # Create indices in Elasticsearch for QA tests.\n    # This is not intended for production use, but can be safely executed in a\n    # production environment without consequence.\n    LOAD_TEST_DATA = 3\n\n\nclass TaskTracker:\n    def __init__(self):\n        self.id_task = {}\n        self.id_action = {}\n        self.id_progress = {}\n        self.id_start_time = {}\n        self.id_finish_time = {}\n\n    def add_task(self, task, task_id, action, progress, finish_time):\n        self._prune_old_tasks()\n        self.id_task[task_id] = task\n        self.id_action[task_id] = action\n        self.id_progress[task_id] = progress\n        self.id_start_time[task_id] = dt.datetime.utcnow().timestamp()\n        self.id_finish_time[task_id] = finish_time\n        return task_id\n\n    def _prune_old_tasks(self):\n        pass\n\n    def list_task_statuses(self):\n        self._prune_old_tasks()\n        results = []\n        for _id, task in self.id_task.items():\n            percent_completed = self.id_progress[_id].value\n            active = task.is_alive()\n            start_time = self.id_start_time[_id]\n            finish_time = self.id_finish_time[_id].value\n            results.append(\n                {\n                    \"task_id\": _id,\n                    \"active\": active,\n                    \"action\": self.id_action[_id],\n                    \"progress\": percent_completed,\n                    \"error\": percent_completed < 100 and not active,\n                    \"start_time\": start_time,\n                    \"finish_time\": finish_time,\n                }\n            )\n        sorted_results = sorted(results, key=lambda x: x[\"finish_time\"])\n\n        to_utc = dt.datetime.utcfromtimestamp\n\n        def render_date(x):\n            return to_utc(x) if x != 0.0 else None\n\n        # Convert date to a readable format\n        for idx, task in enumerate(sorted_results):\n            start_time = task[\"start_time\"]\n            finish_time = task[\"finish_time\"]\n            sorted_results[idx][\"start_time\"] = str(render_date(start_time))\n            sorted_results[idx][\"finish_time\"] = str(render_date(finish_time))\n\n        return sorted_results\n\n\nclass Task(Process):\n    def __init__(\n        self, model, task_type, since_date, progress, task_id, finish_time, callback_url\n    ):\n        Process.__init__(self)\n        self.model = model\n        self.task_type = task_type\n        self.since_date = since_date\n        self.progress = progress\n        self.task_id = task_id\n        self.finish_time = finish_time\n        self.callback_url = callback_url\n\n    def run(self):\n        # Map task types to actions.\n        elasticsearch = elasticsearch_connect()\n        indexer = TableIndexer(\n            elasticsearch, self.model, self.progress, self.finish_time\n        )\n        if self.task_type == TaskTypes.REINDEX:\n            indexer.reindex(self.model)\n        elif self.task_type == TaskTypes.UPDATE_INDEX:\n            indexer.update(self.model, self.since_date)\n        elif self.task_type == TaskTypes.INGEST_UPSTREAM:\n            reload_upstream(self.model)\n            if self.model == \"audio\":\n                reload_upstream(\"audioset\", approach=\"basic\")\n            indexer.reindex(self.model)\n        elif self.task_type == TaskTypes.LOAD_TEST_DATA:\n            indexer.load_test_data(self.model)\n        logging.info(f\"Task {self.task_id} exited.\")\n        if self.callback_url:\n            try:\n                requests.post(self.callback_url)\n            except requests.exceptions.RequestException as e:\n                logging.error(\"Failed to send callback!\")\n                logging.error(e)\n", "path": "ingestion_server/ingestion_server/tasks.py"}], "after_files": [{"content": null, "path": "sample_data/make_sample_pop.py"}, {"content": "\"\"\"\nSimple in-memory tracking of executed tasks.\n\"\"\"\n\nimport datetime as dt\nimport logging\nfrom enum import Enum\nfrom multiprocessing import Process\n\nimport requests\n\nfrom ingestion_server.indexer import TableIndexer, elasticsearch_connect\nfrom ingestion_server.ingest import reload_upstream\n\n\nclass TaskTypes(Enum):\n    # Completely reindex all data for a given model.\n    REINDEX = 0\n    # Reindex updates to a model from the database since a certain date.\n    UPDATE_INDEX = 1\n    # Download the latest copy of the data from the upstream database, then\n    # completely reindex the newly imported data.\n    INGEST_UPSTREAM = 2\n    # Create indices in Elasticsearch for QA tests.\n    # This is not intended for production use, but can be safely executed in a\n    # production environment without consequence.\n    LOAD_TEST_DATA = 3\n\n\nclass TaskTracker:\n    def __init__(self):\n        self.id_task = {}\n        self.id_action = {}\n        self.id_progress = {}\n        self.id_start_time = {}\n        self.id_finish_time = {}\n\n    def add_task(self, task, task_id, action, progress, finish_time):\n        self._prune_old_tasks()\n        self.id_task[task_id] = task\n        self.id_action[task_id] = action\n        self.id_progress[task_id] = progress\n        self.id_start_time[task_id] = dt.datetime.utcnow().timestamp()\n        self.id_finish_time[task_id] = finish_time\n        return task_id\n\n    def _prune_old_tasks(self):\n        pass\n\n    def list_task_statuses(self):\n        self._prune_old_tasks()\n        results = []\n        for _id, task in self.id_task.items():\n            percent_completed = self.id_progress[_id].value\n            active = task.is_alive()\n            start_time = self.id_start_time[_id]\n            finish_time = self.id_finish_time[_id].value\n            results.append(\n                {\n                    \"task_id\": _id,\n                    \"active\": active,\n                    \"action\": self.id_action[_id],\n                    \"progress\": percent_completed,\n                    \"error\": percent_completed < 100 and not active,\n                    \"start_time\": start_time,\n                    \"finish_time\": finish_time,\n                }\n            )\n        sorted_results = sorted(results, key=lambda x: x[\"finish_time\"])\n\n        to_utc = dt.datetime.utcfromtimestamp\n\n        def render_date(x):\n            return to_utc(x) if x != 0.0 else None\n\n        # Convert date to a readable format\n        for idx, task in enumerate(sorted_results):\n            start_time = task[\"start_time\"]\n            finish_time = task[\"finish_time\"]\n            sorted_results[idx][\"start_time\"] = str(render_date(start_time))\n            sorted_results[idx][\"finish_time\"] = str(render_date(finish_time))\n\n        return sorted_results\n\n\nclass Task(Process):\n    def __init__(\n        self, model, task_type, since_date, progress, task_id, finish_time, callback_url\n    ):\n        Process.__init__(self)\n        self.model = model\n        self.task_type = task_type\n        self.since_date = since_date\n        self.progress = progress\n        self.task_id = task_id\n        self.finish_time = finish_time\n        self.callback_url = callback_url\n\n    def run(self):\n        # Map task types to actions.\n        elasticsearch = elasticsearch_connect()\n        indexer = TableIndexer(\n            elasticsearch, self.model, self.progress, self.finish_time\n        )\n        if self.task_type == TaskTypes.REINDEX:\n            indexer.reindex(self.model)\n        elif self.task_type == TaskTypes.UPDATE_INDEX:\n            indexer.update(self.model, self.since_date)\n        elif self.task_type == TaskTypes.INGEST_UPSTREAM:\n            reload_upstream(self.model)\n            if self.model == \"audio\":\n                reload_upstream(\"audioset\", approach=\"basic\")\n            indexer.reindex(self.model)\n        elif self.task_type == TaskTypes.LOAD_TEST_DATA:\n            indexer.load_test_data(self.model)\n        logging.info(f\"Task {self.task_id} exited.\")\n        if self.callback_url:\n            try:\n                logging.info(\"Sending callback request\")\n                res = requests.post(self.callback_url)\n                logging.info(f\"Response: {res.text}\")\n            except requests.exceptions.RequestException as e:\n                logging.error(\"Failed to send callback!\")\n                logging.error(e)\n", "path": "ingestion_server/ingestion_server/tasks.py"}]}