+++ title = "Step 3: Test Our Pipeline" chapter = false weight = 21 +++ In this section we will push a code change to our GitHub to demonstrate how CodePipeline orchestrates our deployment pipeline. We will make the following change to trigger our pipeline to run. Let's go ahead and add a route to our Flask application that returns "Hello World!" to demonstrate that our pipeline works. We will add a route named /test to our Flask application that returns "Hello World!" and the following code changes should be made to backend/hello.py ``` import os from flask import Flask import mysql.connector import uuid import os class DBManager: def __init__(self): password_file = os.environ.get('MY_SQL_PASSWORD_FILE','/run/secrets/db-password') pf = open(password_file, 'r') database = os.environ.get('MY_SQL_DATABASE','example') host = os.environ.get('MY_SQL_HOST',"db") user = os.environ.get('MY_SQL_USER',"root") self.connection = mysql.connector.connect( user=user, password=pf.read(), host=host, # name of the mysql service as set in the docker-compose file database=database, auth_plugin='mysql_native_password' ) pf.close() self.cursor = self.connection.cursor() def populate_db(self): self.cursor.execute('DROP TABLE IF EXISTS blog') self.cursor.execute('CREATE TABLE blog (id INT NOT NULL AUTO_INCREMENT PRIMARY KEY, title VARCHAR(256))') self.cursor.execute("insert into blog values(NULL,'initial load')") self.connection.commit() def insert_records(self,id,name): data = (id,str(name)) query = ( "INSERT INTO blog(id,title)" "VALUES (%s, %s)" ) self.cursor.execute(query,data) self.connection.commit() def query_titles(self): self.cursor.execute('SELECT title FROM blog') rec = [] for c in self.cursor: rec.append(c[0]) return rec server = Flask(__name__) conn = None @server.route('/') def listName(): global conn if not conn: conn = DBManager() conn.populate_db() #conn.insert_records() rec = conn.query_titles() response = '' for c in rec: response = response + '