We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
c7n_kube - tests spin up threads without closing them
which causes test runner to hang if executing without xdist

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `tools/c7n_kube/c7n_kube/server.py`
Content:
```
1 # Copyright The Cloud Custodian Authors.
2 # SPDX-License-Identifier: Apache-2.0
3 import base64
4 import json
5 import os
6 import http.server
7 
8 from c7n.config import Config
9 from c7n.loader import DirectoryLoader
10 
11 from c7n_kube.utils import evaluate_result
12 from c7n_kube.exceptions import EventNotMatchedException, PolicyNotRunnableException
13 
14 import logging
15 
16 log = logging.getLogger("c7n_kube.server")
17 log.setLevel(logging.DEBUG)
18 
19 
20 class AdmissionControllerServer(http.server.HTTPServer):
21     """
22     Admission Controller Server
23     """
24 
25     def __init__(self, policy_dir, on_exception="warn", *args, **kwargs):
26         self.policy_dir = policy_dir
27         self.on_exception = on_exception
28         self.directory_loader = DirectoryLoader(Config.empty())
29         policy_collection = self.directory_loader.load_directory(os.path.abspath(self.policy_dir))
30         self.policy_collection = policy_collection.filter(modes=["k8s-admission"])
31         log.info(f"Loaded {len(self.policy_collection)} policies")
32         super().__init__(*args, **kwargs)
33 
34 
35 class AdmissionControllerHandler(http.server.BaseHTTPRequestHandler):
36     def run_policies(self, req):
37         failed_policies = []
38         warn_policies = []
39         patches = []
40         for p in self.server.policy_collection.policies:
41             # fail_message and warning_message are set on exception
42             warning_message = None
43             deny_message = None
44             resources = None
45             try:
46                 resources = p.push(req)
47                 action = p.data["mode"].get("on-match", "deny")
48                 result = evaluate_result(action, resources)
49                 if result in (
50                     "allow",
51                     "warn",
52                 ):
53                     verb = "allowing"
54                 else:
55                     verb = "denying"
56 
57                 log.info(f"{verb} admission because on-match:{action}, matched:{len(resources)}")
58             except (
59                 PolicyNotRunnableException,
60                 EventNotMatchedException,
61             ):
62                 result = "allow"
63                 resources = []
64             except Exception as e:
65                 # if a policy fails we simply warn
66                 result = self.server.on_exception
67                 if result == "warn":
68                     warning_message = f"Error in executing policy: {str(e)}"
69                 if result == "deny":
70                     deny_message = f"Error in executing policy: {str(e)}"
71 
72             if result == "deny":
73                 failed_policies.append(
74                     {
75                         "name": p.name,
76                         "description": deny_message or p.data.get("description", ""),
77                     }
78                 )
79             if result == "warn":
80                 warn_policies.append(
81                     {
82                         "name": p.name,
83                         "description": warning_message or p.data.get("description", ""),
84                     }
85                 )
86             if resources:
87                 patches.extend(resources[0].get("c7n:patches", []))
88         return failed_policies, warn_policies, patches
89 
90     def get_request_body(self):
91         token = self.rfile.read(int(self.headers["Content-length"]))
92         res = token.decode("utf-8")
93         return res
94 
95     def do_GET(self):
96         """
97         Returns application/json list of your policies
98         """
99         self.send_response(200)
100         self.end_headers()
101         result = []
102         for p in self.server.policy_collection.policies:
103             result.append(p.data)
104         self.wfile.write(json.dumps(result).encode("utf-8"))
105 
106     def do_POST(self):
107         """
108         Entrypoint for kubernetes webhook
109         """
110         req = self.get_request_body()
111         log.info(req)
112         try:
113             req = json.loads(req)
114         except Exception as e:
115             self.send_response(400)
116             self.send_header("Content-Type", "application/json")
117             self.end_headers()
118             self.wfile.write(json.dumps({"error": str(e)}).encode("utf-8"))
119             return
120 
121         failed_policies, warn_policies, patches = self.run_policies(req)
122 
123         self.send_response(200)
124         self.send_header("Content-Type", "application/json")
125         self.end_headers()
126 
127         if patches:
128             patches = base64.b64encode(json.dumps(patches).encode("utf-8")).decode()
129 
130         response = self.create_admission_response(
131             uid=req["request"]["uid"],
132             failed_policies=failed_policies,
133             warn_policies=warn_policies,
134             patches=patches,
135         )
136         log.info(response)
137         self.wfile.write(response.encode("utf-8"))
138 
139     def create_admission_response(
140         self, uid, failed_policies=None, warn_policies=None, patches=None
141     ):
142         code = 200 if len(failed_policies) == 0 else 400
143         message = "OK"
144         warnings = []
145         if failed_policies:
146             message = f"Failed admission due to policies:{json.dumps(failed_policies)}"
147         if warn_policies:
148             for p in warn_policies:
149                 warnings.append(f"{p['name']}:{p['description']}")
150 
151         response = {
152             "apiVersion": "admission.k8s.io/v1",
153             "kind": "AdmissionReview",
154             "response": {
155                 "allowed": False if failed_policies else True,
156                 "warnings": warnings,
157                 "uid": uid,
158                 "status": {"code": code, "message": message},
159             },
160         }
161 
162         if patches:
163             patch = {"patchType": "JSONPatch", "patch": patches}
164             response["response"].update(patch)
165         return json.dumps(response)
166 
167 
168 def init(
169     host,
170     port,
171     policy_dir,
172     on_exception="warn",
173     serve_forever=True,
174     *,
175     cert_path=None,
176     cert_key_path=None,
177     ca_cert_path=None,
178 ):
179     use_tls = any((cert_path, cert_key_path))
180     if use_tls and not (cert_path and cert_key_path):
181         raise Exception(
182             "must configure both a certificate and a key to enable TLS",
183         )
184 
185     server = AdmissionControllerServer(
186         server_address=(host, port),
187         RequestHandlerClass=AdmissionControllerHandler,
188         policy_dir=policy_dir,
189         on_exception=on_exception,
190     )
191     if use_tls:
192         import ssl
193 
194         server.socket = ssl.wrap_socket(
195             server.socket,
196             server_side=True,
197             certfile=cert_path,
198             keyfile=cert_key_path,
199             ca_certs=ca_cert_path,
200         )
201 
202     log.info(f"Serving at http{'s' if use_tls else ''}://{host}:{port}")
203     while True:
204         server.serve_forever()
205         # for testing purposes
206         if not serve_forever:
207             break
208
```
--- 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()

```

diff --git a/tools/c7n_kube/c7n_kube/server.py b/tools/c7n_kube/c7n_kube/server.py
--- a/tools/c7n_kube/c7n_kube/server.py
+++ b/tools/c7n_kube/c7n_kube/server.py
@@ -1,9 +1,10 @@
 # Copyright The Cloud Custodian Authors.
 # SPDX-License-Identifier: Apache-2.0
 import base64
+import http.server
 import json
 import os
-import http.server
+import tempfile
 
 from c7n.config import Config
 from c7n.loader import DirectoryLoader
@@ -25,7 +26,8 @@
     def __init__(self, policy_dir, on_exception="warn", *args, **kwargs):
         self.policy_dir = policy_dir
         self.on_exception = on_exception
-        self.directory_loader = DirectoryLoader(Config.empty())
+        temp_dir = tempfile.TemporaryDirectory()
+        self.directory_loader = DirectoryLoader(Config.empty(output_dir=temp_dir.name))
         policy_collection = self.directory_loader.load_directory(os.path.abspath(self.policy_dir))
         self.policy_collection = policy_collection.filter(modes=["k8s-admission"])
         log.info(f"Loaded {len(self.policy_collection)} policies")

{"golden_diff": "diff --git a/tools/c7n_kube/c7n_kube/server.py b/tools/c7n_kube/c7n_kube/server.py\n--- a/tools/c7n_kube/c7n_kube/server.py\n+++ b/tools/c7n_kube/c7n_kube/server.py\n@@ -1,9 +1,10 @@\n # Copyright The Cloud Custodian Authors.\n # SPDX-License-Identifier: Apache-2.0\n import base64\n+import http.server\n import json\n import os\n-import http.server\n+import tempfile\n \n from c7n.config import Config\n from c7n.loader import DirectoryLoader\n@@ -25,7 +26,8 @@\n     def __init__(self, policy_dir, on_exception=\"warn\", *args, **kwargs):\n         self.policy_dir = policy_dir\n         self.on_exception = on_exception\n-        self.directory_loader = DirectoryLoader(Config.empty())\n+        temp_dir = tempfile.TemporaryDirectory()\n+        self.directory_loader = DirectoryLoader(Config.empty(output_dir=temp_dir.name))\n         policy_collection = self.directory_loader.load_directory(os.path.abspath(self.policy_dir))\n         self.policy_collection = policy_collection.filter(modes=[\"k8s-admission\"])\n         log.info(f\"Loaded {len(self.policy_collection)} policies\")\n", "issue": "c7n_kube - tests spin up threads without closing them\nwhich causes test runner to hang if executing without xdist\n", "before_files": [{"content": "# Copyright The Cloud Custodian Authors.\n# SPDX-License-Identifier: Apache-2.0\nimport base64\nimport json\nimport os\nimport http.server\n\nfrom c7n.config import Config\nfrom c7n.loader import DirectoryLoader\n\nfrom c7n_kube.utils import evaluate_result\nfrom c7n_kube.exceptions import EventNotMatchedException, PolicyNotRunnableException\n\nimport logging\n\nlog = logging.getLogger(\"c7n_kube.server\")\nlog.setLevel(logging.DEBUG)\n\n\nclass AdmissionControllerServer(http.server.HTTPServer):\n    \"\"\"\n    Admission Controller Server\n    \"\"\"\n\n    def __init__(self, policy_dir, on_exception=\"warn\", *args, **kwargs):\n        self.policy_dir = policy_dir\n        self.on_exception = on_exception\n        self.directory_loader = DirectoryLoader(Config.empty())\n        policy_collection = self.directory_loader.load_directory(os.path.abspath(self.policy_dir))\n        self.policy_collection = policy_collection.filter(modes=[\"k8s-admission\"])\n        log.info(f\"Loaded {len(self.policy_collection)} policies\")\n        super().__init__(*args, **kwargs)\n\n\nclass AdmissionControllerHandler(http.server.BaseHTTPRequestHandler):\n    def run_policies(self, req):\n        failed_policies = []\n        warn_policies = []\n        patches = []\n        for p in self.server.policy_collection.policies:\n            # fail_message and warning_message are set on exception\n            warning_message = None\n            deny_message = None\n            resources = None\n            try:\n                resources = p.push(req)\n                action = p.data[\"mode\"].get(\"on-match\", \"deny\")\n                result = evaluate_result(action, resources)\n                if result in (\n                    \"allow\",\n                    \"warn\",\n                ):\n                    verb = \"allowing\"\n                else:\n                    verb = \"denying\"\n\n                log.info(f\"{verb} admission because on-match:{action}, matched:{len(resources)}\")\n            except (\n                PolicyNotRunnableException,\n                EventNotMatchedException,\n            ):\n                result = \"allow\"\n                resources = []\n            except Exception as e:\n                # if a policy fails we simply warn\n                result = self.server.on_exception\n                if result == \"warn\":\n                    warning_message = f\"Error in executing policy: {str(e)}\"\n                if result == \"deny\":\n                    deny_message = f\"Error in executing policy: {str(e)}\"\n\n            if result == \"deny\":\n                failed_policies.append(\n                    {\n                        \"name\": p.name,\n                        \"description\": deny_message or p.data.get(\"description\", \"\"),\n                    }\n                )\n            if result == \"warn\":\n                warn_policies.append(\n                    {\n                        \"name\": p.name,\n                        \"description\": warning_message or p.data.get(\"description\", \"\"),\n                    }\n                )\n            if resources:\n                patches.extend(resources[0].get(\"c7n:patches\", []))\n        return failed_policies, warn_policies, patches\n\n    def get_request_body(self):\n        token = self.rfile.read(int(self.headers[\"Content-length\"]))\n        res = token.decode(\"utf-8\")\n        return res\n\n    def do_GET(self):\n        \"\"\"\n        Returns application/json list of your policies\n        \"\"\"\n        self.send_response(200)\n        self.end_headers()\n        result = []\n        for p in self.server.policy_collection.policies:\n            result.append(p.data)\n        self.wfile.write(json.dumps(result).encode(\"utf-8\"))\n\n    def do_POST(self):\n        \"\"\"\n        Entrypoint for kubernetes webhook\n        \"\"\"\n        req = self.get_request_body()\n        log.info(req)\n        try:\n            req = json.loads(req)\n        except Exception as e:\n            self.send_response(400)\n            self.send_header(\"Content-Type\", \"application/json\")\n            self.end_headers()\n            self.wfile.write(json.dumps({\"error\": str(e)}).encode(\"utf-8\"))\n            return\n\n        failed_policies, warn_policies, patches = self.run_policies(req)\n\n        self.send_response(200)\n        self.send_header(\"Content-Type\", \"application/json\")\n        self.end_headers()\n\n        if patches:\n            patches = base64.b64encode(json.dumps(patches).encode(\"utf-8\")).decode()\n\n        response = self.create_admission_response(\n            uid=req[\"request\"][\"uid\"],\n            failed_policies=failed_policies,\n            warn_policies=warn_policies,\n            patches=patches,\n        )\n        log.info(response)\n        self.wfile.write(response.encode(\"utf-8\"))\n\n    def create_admission_response(\n        self, uid, failed_policies=None, warn_policies=None, patches=None\n    ):\n        code = 200 if len(failed_policies) == 0 else 400\n        message = \"OK\"\n        warnings = []\n        if failed_policies:\n            message = f\"Failed admission due to policies:{json.dumps(failed_policies)}\"\n        if warn_policies:\n            for p in warn_policies:\n                warnings.append(f\"{p['name']}:{p['description']}\")\n\n        response = {\n            \"apiVersion\": \"admission.k8s.io/v1\",\n            \"kind\": \"AdmissionReview\",\n            \"response\": {\n                \"allowed\": False if failed_policies else True,\n                \"warnings\": warnings,\n                \"uid\": uid,\n                \"status\": {\"code\": code, \"message\": message},\n            },\n        }\n\n        if patches:\n            patch = {\"patchType\": \"JSONPatch\", \"patch\": patches}\n            response[\"response\"].update(patch)\n        return json.dumps(response)\n\n\ndef init(\n    host,\n    port,\n    policy_dir,\n    on_exception=\"warn\",\n    serve_forever=True,\n    *,\n    cert_path=None,\n    cert_key_path=None,\n    ca_cert_path=None,\n):\n    use_tls = any((cert_path, cert_key_path))\n    if use_tls and not (cert_path and cert_key_path):\n        raise Exception(\n            \"must configure both a certificate and a key to enable TLS\",\n        )\n\n    server = AdmissionControllerServer(\n        server_address=(host, port),\n        RequestHandlerClass=AdmissionControllerHandler,\n        policy_dir=policy_dir,\n        on_exception=on_exception,\n    )\n    if use_tls:\n        import ssl\n\n        server.socket = ssl.wrap_socket(\n            server.socket,\n            server_side=True,\n            certfile=cert_path,\n            keyfile=cert_key_path,\n            ca_certs=ca_cert_path,\n        )\n\n    log.info(f\"Serving at http{'s' if use_tls else ''}://{host}:{port}\")\n    while True:\n        server.serve_forever()\n        # for testing purposes\n        if not serve_forever:\n            break\n", "path": "tools/c7n_kube/c7n_kube/server.py"}], "after_files": [{"content": "# Copyright The Cloud Custodian Authors.\n# SPDX-License-Identifier: Apache-2.0\nimport base64\nimport http.server\nimport json\nimport os\nimport tempfile\n\nfrom c7n.config import Config\nfrom c7n.loader import DirectoryLoader\n\nfrom c7n_kube.utils import evaluate_result\nfrom c7n_kube.exceptions import EventNotMatchedException, PolicyNotRunnableException\n\nimport logging\n\nlog = logging.getLogger(\"c7n_kube.server\")\nlog.setLevel(logging.DEBUG)\n\n\nclass AdmissionControllerServer(http.server.HTTPServer):\n    \"\"\"\n    Admission Controller Server\n    \"\"\"\n\n    def __init__(self, policy_dir, on_exception=\"warn\", *args, **kwargs):\n        self.policy_dir = policy_dir\n        self.on_exception = on_exception\n        temp_dir = tempfile.TemporaryDirectory()\n        self.directory_loader = DirectoryLoader(Config.empty(output_dir=temp_dir.name))\n        policy_collection = self.directory_loader.load_directory(os.path.abspath(self.policy_dir))\n        self.policy_collection = policy_collection.filter(modes=[\"k8s-admission\"])\n        log.info(f\"Loaded {len(self.policy_collection)} policies\")\n        super().__init__(*args, **kwargs)\n\n\nclass AdmissionControllerHandler(http.server.BaseHTTPRequestHandler):\n    def run_policies(self, req):\n        failed_policies = []\n        warn_policies = []\n        patches = []\n        for p in self.server.policy_collection.policies:\n            # fail_message and warning_message are set on exception\n            warning_message = None\n            deny_message = None\n            resources = None\n            try:\n                resources = p.push(req)\n                action = p.data[\"mode\"].get(\"on-match\", \"deny\")\n                result = evaluate_result(action, resources)\n                if result in (\n                    \"allow\",\n                    \"warn\",\n                ):\n                    verb = \"allowing\"\n                else:\n                    verb = \"denying\"\n\n                log.info(f\"{verb} admission because on-match:{action}, matched:{len(resources)}\")\n            except (\n                PolicyNotRunnableException,\n                EventNotMatchedException,\n            ):\n                result = \"allow\"\n                resources = []\n            except Exception as e:\n                # if a policy fails we simply warn\n                result = self.server.on_exception\n                if result == \"warn\":\n                    warning_message = f\"Error in executing policy: {str(e)}\"\n                if result == \"deny\":\n                    deny_message = f\"Error in executing policy: {str(e)}\"\n\n            if result == \"deny\":\n                failed_policies.append(\n                    {\n                        \"name\": p.name,\n                        \"description\": deny_message or p.data.get(\"description\", \"\"),\n                    }\n                )\n            if result == \"warn\":\n                warn_policies.append(\n                    {\n                        \"name\": p.name,\n                        \"description\": warning_message or p.data.get(\"description\", \"\"),\n                    }\n                )\n            if resources:\n                patches.extend(resources[0].get(\"c7n:patches\", []))\n        return failed_policies, warn_policies, patches\n\n    def get_request_body(self):\n        token = self.rfile.read(int(self.headers[\"Content-length\"]))\n        res = token.decode(\"utf-8\")\n        return res\n\n    def do_GET(self):\n        \"\"\"\n        Returns application/json list of your policies\n        \"\"\"\n        self.send_response(200)\n        self.end_headers()\n        result = []\n        for p in self.server.policy_collection.policies:\n            result.append(p.data)\n        self.wfile.write(json.dumps(result).encode(\"utf-8\"))\n\n    def do_POST(self):\n        \"\"\"\n        Entrypoint for kubernetes webhook\n        \"\"\"\n        req = self.get_request_body()\n        log.info(req)\n        try:\n            req = json.loads(req)\n        except Exception as e:\n            self.send_response(400)\n            self.send_header(\"Content-Type\", \"application/json\")\n            self.end_headers()\n            self.wfile.write(json.dumps({\"error\": str(e)}).encode(\"utf-8\"))\n            return\n\n        failed_policies, warn_policies, patches = self.run_policies(req)\n\n        self.send_response(200)\n        self.send_header(\"Content-Type\", \"application/json\")\n        self.end_headers()\n\n        if patches:\n            patches = base64.b64encode(json.dumps(patches).encode(\"utf-8\")).decode()\n\n        response = self.create_admission_response(\n            uid=req[\"request\"][\"uid\"],\n            failed_policies=failed_policies,\n            warn_policies=warn_policies,\n            patches=patches,\n        )\n        log.info(response)\n        self.wfile.write(response.encode(\"utf-8\"))\n\n    def create_admission_response(\n        self, uid, failed_policies=None, warn_policies=None, patches=None\n    ):\n        code = 200 if len(failed_policies) == 0 else 400\n        message = \"OK\"\n        warnings = []\n        if failed_policies:\n            message = f\"Failed admission due to policies:{json.dumps(failed_policies)}\"\n        if warn_policies:\n            for p in warn_policies:\n                warnings.append(f\"{p['name']}:{p['description']}\")\n\n        response = {\n            \"apiVersion\": \"admission.k8s.io/v1\",\n            \"kind\": \"AdmissionReview\",\n            \"response\": {\n                \"allowed\": False if failed_policies else True,\n                \"warnings\": warnings,\n                \"uid\": uid,\n                \"status\": {\"code\": code, \"message\": message},\n            },\n        }\n\n        if patches:\n            patch = {\"patchType\": \"JSONPatch\", \"patch\": patches}\n            response[\"response\"].update(patch)\n        return json.dumps(response)\n\n\ndef init(\n    host,\n    port,\n    policy_dir,\n    on_exception=\"warn\",\n    serve_forever=True,\n    *,\n    cert_path=None,\n    cert_key_path=None,\n    ca_cert_path=None,\n):\n    use_tls = any((cert_path, cert_key_path))\n    if use_tls and not (cert_path and cert_key_path):\n        raise Exception(\n            \"must configure both a certificate and a key to enable TLS\",\n        )\n\n    server = AdmissionControllerServer(\n        server_address=(host, port),\n        RequestHandlerClass=AdmissionControllerHandler,\n        policy_dir=policy_dir,\n        on_exception=on_exception,\n    )\n    if use_tls:\n        import ssl\n\n        server.socket = ssl.wrap_socket(\n            server.socket,\n            server_side=True,\n            certfile=cert_path,\n            keyfile=cert_key_path,\n            ca_certs=ca_cert_path,\n        )\n\n    log.info(f\"Serving at http{'s' if use_tls else ''}://{host}:{port}\")\n    while True:\n        server.serve_forever()\n        # for testing purposes\n        if not serve_forever:\n            break\n", "path": "tools/c7n_kube/c7n_kube/server.py"}]}