Repository: getmoto/moto (python)

Issue:
Unable to easily mock Athena result CSV
We are able to mock Athena query results via:

```python
   resp = requests.post(
        "http://motoapi.amazonaws.com/moto-api/static/athena/query-results",
        json=query_results,
    )
    assert resp.status_code == 201
``` 

But there is no way to mock the output of `get_execution_results`. this poses a few problems:

1. unable to have any control over request status (in cases where you want to test that retries are happening for instance)
2. unable to have any control over the output location

I am more interested in the latter. This would be a lot more useful if we could somehow assign results via a CSV, and have the `query-results` mocked from that. But in the simplest case, being able to control that datapoint to point to a file we have put to a mock bucket
would go a long way in the case where the code under test needs to access the file directly. 

Here is an example of how I had to hack this into my tests:

```python
from moto.athena import athena_backends
from moto.athena.models import Execution
from moto.core import DEFAULT_ACCOUNT_ID

@pytest.fixture
@mock_aws
def mock_athena_results():
    # place the mocked athena results in "s3"
    s3 = boto3.client("s3", region_name=os.getenv("AWS_DEFAULT_REGION"))
    bucket_name = 'test-bucket'
    key = f"query_results/output.csv"
  
    with open('some/csv/file.csv', "r") as f:
        contents = f.read()
        s3.put_object(Bucket=bucket_name, Key=key, Body=contents)

    # Hack moto to mock the execution results to have the correct path to our csv results
    exec = Execution(
        query="SELECT * FROM notused",
        config={
            "OutputLocation": f"s3://test-bucket/query_results/output.csv"
        },
        context={
            'Database': 'default'
        },
        workgroup={},
        execution_parameters={}
    )
    # moto tries to mock the file location to _look_ correct, but we need it to be an actual result
    exec.config['OutputLocation'] = f"s3://test-bucket/query_results/output.csv"

    # This is a hack to get moto to return the correct execution object by having a factory that
    # pretends to be a list of executions
    class MockExecutionFactory:
        def __get__(self, instance, owner):
            return [
                exec
            ]

        def __set__(self, instance, value):
            pass

        def __setitem__(self, key, value):
            pass

        def __getitem__(self, key):
            return exec

    athena_backends[DEFAULT_ACCOUNT_ID]["us-east-1"].executions = MockExecutionFactory()
    return
```

The above works, but its pretty janky. I suppose this is more of a feature request than anything. Or even better yet, having a way to control execution id (not with seeding, thats still too unpredictable and fragile) such that it can be explicitly set. 



Pull request description:
Athena: Store QueryResults in S3
Fixes #7702 

Patch:
diff --git a/docs/docs/services/athena.rst b/docs/docs/services/athena.rst
index 12145c85a..c90ccaf82 100644
--- a/docs/docs/services/athena.rst
+++ b/docs/docs/services/athena.rst
@@ -88,6 +88,8 @@ athena
 
         .. note:: The exact QueryExecutionId is not relevant here, but will likely be whatever value is returned by start_query_execution
 
+        Query results will also be stored in the S3 output location (in CSV format).
+
         
 
 - [ ] get_query_runtime_statistics
diff --git a/moto/athena/models.py b/moto/athena/models.py
index 93a563ae2..1ca9d35bd 100644
--- a/moto/athena/models.py
+++ b/moto/athena/models.py
@@ -5,6 +5,8 @@ from typing import Any, Dict, List, Optional
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
 from moto.moto_api._internal import mock_random
+from moto.s3.models import s3_backends
+from moto.s3.utils import bucket_and_name_from_url
 from moto.utilities.paginator import paginate
 from moto.utilities.utils import get_partition
 
@@ -232,8 +234,17 @@ class AthenaBackend(BaseBackend):
             execution_parameters=execution_parameters,
         )
         self.executions[execution.id] = execution
+
+        self._store_predefined_query_results(execution.id)
+
         return execution.id
 
+    def _store_predefined_query_results(self, exec_id: str) -> None:
+        if exec_id not in self.query_results and self.query_results_queue:
+            self.query_results[exec_id] = self.query_results_queue.pop(0)
+
+            self._store_query_result_in_s3(exec_id)
+
     def get_query_execution(self, exec_id: str) -> Execution:
         return self.executions[exec_id]
 
@@ -285,9 +296,11 @@ class AthenaBackend(BaseBackend):
 
         .. note:: The exact QueryExecutionId is not relevant here, but will likely be whatever value is returned by start_query_execution
 
+        Query results will also be stored in the S3 output location (in CSV format).
+
         """
-        if exec_id not in self.query_results and self.query_results_queue:
-            self.query_results[exec_id] = self.query_results_queue.pop(0)
+        self._store_predefined_query_results(exec_id)
+
         results = (
             self.query_results[exec_id]
             if exec_id in self.query_results
@@ -295,6 +308,31 @@ class AthenaBackend(BaseBackend):
         )
         return results
 
+    def _store_query_result_in_s3(self, exec_id: str) -> None:
+        try:
+            output_location = self.executions[exec_id].config["OutputLocation"]
+            bucket, key = bucket_and_name_from_url(output_location)
+
+            query_result = ""
+            for row in self.query_results[exec_id].rows:
+                query_result += ",".join(
+                    [
+                        f"\"{r['VarCharValue']}\"" if "VarCharValue" in r else ""
+                        for r in row["Data"]
+                    ]
+                )
+                query_result += "\n"
+
+            s3_backends[self.account_id][self.partition].put_object(
+                bucket_name=bucket,  # type: ignore
+                key_name=key,  # type: ignore
+                value=query_result.encode("utf-8"),
+            )
+        except:  # noqa
+            # Execution may not exist
+            # OutputLocation may not exist
+            pass
+
     def stop_query_execution(self, exec_id: str) -> None:
         execution = self.executions[exec_id]
         execution.status = "CANCELLED"
diff --git a/moto/moto_proxy/certificate_creator.py b/moto/moto_proxy/certificate_creator.py
index c5699a0b4..db0c29dbe 100644
--- a/moto/moto_proxy/certificate_creator.py
+++ b/moto/moto_proxy/certificate_creator.py
@@ -64,10 +64,13 @@ class CertificateCreator:
             # All requests that match this wildcard certificate will reuse it
 
             wildcard_name = f"*.{'.'.join(full_name.split('.')[1:])}"
-            server_csr = f"{self.certdir.rstrip('/')}/{wildcard_name}.csr"
+            # In Linux we could use asterisk as part of the filename, but not on Windows
+            # So we need to pick a different (unique) filename that works everywhere
+            wildcard_filename = f"_.{'.'.join(full_name.split('.')[1:])}"
+            server_csr = f"{self.certdir.rstrip('/')}/{wildcard_filename}.csr"
 
             # Verify if the certificate already exists
-            certpath = f"{self.certdir.rstrip('/')}/{wildcard_name}.crt"
+            certpath = f"{self.certdir.rstrip('/')}/{wildcard_filename}.crt"
             if not os.path.isfile(certpath):
                 # Create a Config-file that contains the wildcard-name
                 with open(f"{self.certdir.rstrip('/')}/req.conf.tmpl", "r") as f:
@@ -77,12 +80,12 @@ class CertificateCreator:
                     "{{wildcard_name}}", wildcard_name
                 )
                 config_template_name = (
-                    f"{self.certdir.rstrip('/')}/{wildcard_name}.conf"
+                    f"{self.certdir.rstrip('/')}/{wildcard_filename}.conf"
                 )
                 with open(config_template_name, "w") as f:
                     f.write(config_template)
 
-                # Create an Certificate Signing Request
+                # Create a Certificate Signing Request
                 #
                 subject = f"/CN={full_name}"[0:64]
                 commands = [


Test patch:
diff --git a/.github/workflows/tests_proxymode.yml b/.github/workflows/tests_proxymode.yml
index af7c799db..9abf533be 100644
--- a/.github/workflows/tests_proxymode.yml
+++ b/.github/workflows/tests_proxymode.yml
@@ -3,11 +3,17 @@ on: [workflow_call]
 
 jobs:
   test:
-    runs-on: ubuntu-latest
     strategy:
       fail-fast: false
       matrix:
+        os: ["ubuntu-latest", "windows-latest"]
         python-version: ["3.10", "3.11", "3.12"]
+        include:
+          - os: ubuntu-latest
+            path: ~/.cache/pip
+          - os: windows-latest
+            path: ~\AppData\Local\pip\Cache
+    runs-on: ${{ matrix.os }}
 
     steps:
     - uses: actions/checkout@v4
@@ -24,8 +30,10 @@ jobs:
     - name: pip cache
       uses: actions/cache@v4
       with:
-        path: ${{ steps.pip-cache.outputs.dir }}
-        key: pip-${{ matrix.python-version }}-${{ hashFiles('**/setup.cfg') }}
+        path: ${{ matrix.path }}
+        key: pip-${{ runner.os }}-${{ matrix.python-version }}-${{ hashFiles('**/setup.cfg') }}
+        restore-keys: |
+          pip-${{ runner.os }}-${{ matrix.python-version }}-
     - name: Update pip
       run: |
         python -m pip install --upgrade pip
@@ -33,23 +41,38 @@ jobs:
       run: |
         pip install -r requirements-tests.txt
         pip install .[all,server]
-    - name: Start MotoProxy
+    - name: Start MotoProxy on Linux
+      if: ${{ matrix.os == 'ubuntu-latest' }}
       run: |
-        moto_proxy -h > moto_proxy.log
         moto_proxy -H 0.0.0.0 -v > moto_proxy.log &
-    - name: Test ProxyMode
+    - name: Test ProxyMode on Linux
+      if: ${{ matrix.os == 'ubuntu-latest' }}
       env:
         TEST_PROXY_MODE: ${{ true }}
       run: |
         pytest -sv tests/test_acmpca tests/test_athena tests/test_awslambda tests/test_apigateway tests/test_core/test_proxy.py tests/test_s3
+    - name: Test ProxyMode on Windows
+      if: ${{ matrix.os == 'windows-latest' }}
+      shell: pwsh
+      env:
+        TEST_PROXY_MODE: ${{ true }}
+      # We should start moto_proxy in a different step
+      # But I can't keep the process alive, across different steps
+      # It does run in the background - but it looks like the process is killed at the end of the step
+      # Starting the proxy in the same step has the same outcome though, even if it's not pretty
+      #
+      # This step does not test everything, because it takes ages on Windows for some reason
+      run: |
+        moto_proxy -v &
+        pytest -sv tests/test_acmpca tests/test_core/test_proxy.py
     - name: "Stop MotoProxy"
-      if: always()
+      if: ${{ matrix.os == 'ubuntu-latest' }}
       run: |
         kill $(lsof -t -i:5005)
     - name: Archive Proxy logs
       if: always()
       uses: actions/upload-artifact@v4
       with:
-        name: motoproxy-${{ matrix.python-version }}
+        name: motoproxy-${{ matrix.os }}-${{ matrix.python-version }}
         path: |
           moto_proxy.log
\ No newline at end of file
diff --git a/tests/test_athena/__init__.py b/tests/test_athena/__init__.py
index 08a1c1568..1822c621a 100644
--- a/tests/test_athena/__init__.py
+++ b/tests/test_athena/__init__.py
@@ -1,1 +1,59 @@
-# This file is intentionally left blank.
+import os
+from functools import wraps
+from uuid import uuid4
+
+import boto3
+
+from moto import mock_aws
+from moto.s3.responses import DEFAULT_REGION_NAME
+from tests.test_s3 import empty_bucket
+
+
+def athena_aws_verified(func):
+    """
+    Function that is verified to work against AWS.
+    Can be run against AWS at any time by setting:
+      MOTO_TEST_ALLOW_AWS_REQUEST=true
+
+    If this environment variable is not set, the function runs in a `mock_aws` context.
+
+    This decorator will:
+      - Create a bucket
+      - Run the test and pass the bucket_name as an argument
+      - Delete the objects and the bucket itself
+    """
+
+    @wraps(func)
+    def pagination_wrapper():
+        bucket_name = str(uuid4())
+
+        allow_aws_request = (
+            os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
+        )
+
+        if allow_aws_request:
+            resp = create_bucket_and_test(bucket_name, is_aws=True)
+        else:
+            with mock_aws():
+                resp = create_bucket_and_test(bucket_name, is_aws=False)
+        return resp
+
+    def create_bucket_and_test(bucket_name: str, is_aws: bool):
+        client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+
+        client.create_bucket(Bucket=bucket_name)
+        client.put_bucket_tagging(
+            Bucket=bucket_name,
+            Tagging={"TagSet": [{"Key": "environment", "Value": "moto_tests"}]},
+        )
+        try:
+            resp = func(bucket_name, is_aws)
+        finally:
+            ### CLEANUP ###
+
+            empty_bucket(client, bucket_name)
+            client.delete_bucket(Bucket=bucket_name)
+
+        return resp
+
+    return pagination_wrapper
diff --git a/tests/test_athena/test_athena_integration.py b/tests/test_athena/test_athena_integration.py
new file mode 100644
index 000000000..6af2ffca8
--- /dev/null
+++ b/tests/test_athena/test_athena_integration.py
@@ -0,0 +1,155 @@
+import json
+from unittest import SkipTest
+from uuid import uuid4
+
+import boto3
+import pytest
+import requests
+
+from moto import settings
+from tests.test_athena import athena_aws_verified
+
+DATA = ""
+DATA += json.dumps({"k1": "r1_v1", "k2": "r1_v2", "k3": 1}) + "\n"
+DATA += json.dumps({"k1": "r2_v1", "k2": "r2_v2", "k3": 2}) + "\n"
+DATA += json.dumps({"k1": "r3_v1", "k2": "r3_v2"})
+
+ROWS_DATA = [
+    {"Data": [{"VarCharValue": "k1"}, {"VarCharValue": "k2"}, {"VarCharValue": "k3"}]},
+    {
+        "Data": [
+            {"VarCharValue": "r1_v1"},
+            {"VarCharValue": "r1_v2"},
+            {"VarCharValue": "1"},
+        ]
+    },
+    {
+        "Data": [
+            {"VarCharValue": "r2_v1"},
+            {"VarCharValue": "r2_v2"},
+            {"VarCharValue": "2"},
+        ]
+    },
+    {"Data": [{"VarCharValue": "r3_v1"}, {"VarCharValue": "r3_v2"}, {}]},
+]
+COLUMN_INFO = [
+    {
+        "CatalogName": "hive",
+        "SchemaName": "",
+        "TableName": "",
+        "Name": "k1",
+        "Label": "k1",
+        "Type": "varchar",
+        "Precision": 2147483647,
+        "Scale": 0,
+        "Nullable": "UNKNOWN",
+        "CaseSensitive": True,
+    },
+    {
+        "CatalogName": "hive",
+        "SchemaName": "",
+        "TableName": "",
+        "Name": "k2",
+        "Label": "k2",
+        "Type": "varchar",
+        "Precision": 2147483647,
+        "Scale": 0,
+        "Nullable": "UNKNOWN",
+        "CaseSensitive": True,
+    },
+    {
+        "CatalogName": "hive",
+        "SchemaName": "",
+        "TableName": "",
+        "Name": "k3",
+        "Label": "k3",
+        "Type": "integer",
+        "Precision": 10,
+        "Scale": 0,
+        "Nullable": "UNKNOWN",
+        "CaseSensitive": False,
+    },
+]
+
+
+@athena_aws_verified
+@pytest.mark.aws_verified
+def test_athena_csv_result(bucket_name=None, is_aws: bool = False):
+    if not settings.TEST_DECORATOR_MODE:
+        raise SkipTest("No point in testing this outside of decorators")
+
+    s3 = boto3.client("s3", "us-east-1")
+    s3.put_object(Bucket=bucket_name, Key="input/data1.json", Body=DATA.encode("utf-8"))
+
+    athena = boto3.client("athena", "us-east-1")
+
+    config = {"OutputLocation": f"s3://{bucket_name}/output/"}
+    db_name = f"db_{str(uuid4())[0:6]}"
+
+    athena.start_query_execution(
+        QueryString=f"create database {db_name}", ResultConfiguration=config
+    )
+
+    context = {"Database": db_name}
+    CREATE_TABLE = f"""CREATE EXTERNAL TABLE mytable (
+k1 string,
+k2 string,
+k3 int
+) 
+ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
+STORED AS INPUTFORMAT 'org.apache.hadoop.mapred.TextInputFormat'
+OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat'
+LOCATION 's3://{bucket_name}/input/'; """
+    get_query_results(CREATE_TABLE, athena, config, context)
+
+    if not is_aws:
+        athena_result = {
+            "results": [
+                {
+                    "rows": ROWS_DATA,
+                    "column_info": COLUMN_INFO,
+                }
+            ]
+        }
+        requests.post(
+            "http://motoapi.amazonaws.com/moto-api/static/athena/query-results",
+            json=athena_result,
+        )
+
+    query_exec_id, resp = get_query_results(
+        "SELECT * FROM mytable", athena, config, context
+    )
+    assert resp["ResultSet"] == {
+        "Rows": ROWS_DATA,
+        "ResultSetMetadata": {"ColumnInfo": COLUMN_INFO},
+    }
+
+    file = s3.get_object(Bucket=bucket_name, Key=f"output/{query_exec_id}.csv")
+    assert (
+        file["Body"].read()
+        == b'"k1","k2","k3"\n"r1_v1","r1_v2","1"\n"r2_v1","r2_v2","2"\n"r3_v1","r3_v2",\n'
+    )
+
+    get_query_results("DROP TABLE mytable", athena, config, context)
+
+    get_query_results(f"DROP database {db_name}", athena, config, context)
+
+
+def get_query_results(CREATE_TABLE, athena, config, context):
+    resp = athena.start_query_execution(
+        QueryString=CREATE_TABLE,
+        QueryExecutionContext=context,
+        ResultConfiguration=config,
+    )
+    query_execution_id = resp["QueryExecutionId"]
+
+    status = ""
+    while status not in ["SUCCEEDED", "FAILED"]:
+        resp = athena.get_query_execution(QueryExecutionId=query_execution_id)[
+            "QueryExecution"
+        ]
+
+        status = resp["Status"]["State"]
+
+    resp = athena.get_query_results(QueryExecutionId=query_execution_id)
+    return query_execution_id, resp