test_ingcp.py

#   Copyright 2021 Google LLC
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.
import tftest
import os
import tempfile
import base64
from contextlib import redirect_stdout
from .helpers import fixture_to_pubsub, load_config
import unittest
import logging
import main
import io
from datetime import datetime


class TestIngcp(unittest.TestCase):

    def test_in_gcp(self):
        self._test_in_gcp(True)
        self._test_in_gcp(False)

    def _test_in_gcp(self, legacy=False):
        ALL_TESTS = ['bigquery', 'gcs', 'cai']
        logger = logging.getLogger('test')
        logger.setLevel(logging.DEBUG)

        tf_vars = {
            'project_id':
                os.getenv('PROJECT_ID')
                if os.getenv('PROJECT_ID') else os.getenv('GOOGLE_PROJECT_ID')
        }
        if not tf_vars['project_id'] or tf_vars['project_id'] == '':
            self.skipTest(
                'No PROJECT_ID or GOOGLE_PROJECT_ID environment variable set.')

        tf = tftest.TerraformTest('test/fixtures/gcp')
        tf.setup()
        tfapply = tf.apply(output=True, tf_vars=tf_vars)
        logger.debug('Apply output: %s' % (tfapply))

        tfout = tf.output()
        try:
            params = {}
            sa_key = tempfile.NamedTemporaryFile()
            for k, v in tfout.items():
                if k != 'sa_key':
                    params['%%{%s}' % k.upper()] = v['value']
                else:
                    sa_key.write(base64.b64decode(v['value'].encode()))
                    sa_key.flush()

            os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = sa_key.name

            for test in ALL_TESTS:
                config = None
                if legacy:
                    config = load_config('legacy/%s' % (test), params)
                else:
                    config = load_config(test, params)
                data, context = fixture_to_pubsub(test)
                context.timestamp = datetime.utcnow().strftime(
                    '%Y-%m-%dT%H:%M:%S.%fZ')

                buf = io.StringIO()
                with redirect_stdout(buf):
                    main.decode_and_process(logger, config, data, context)

        except Exception as e:
            tfdestroy = tf.destroy(output=True, tf_vars=tf_vars)
            logger.debug('Destroy output: %s' % (tfdestroy))
            raise (e)


if __name__ == '__main__':
    unittest.main()