@skipIf((PSYCOPG2_VERSION < (2, 7)), 'SQL string composition not available in psycopg2<2.7')
def test_composed_query_identifier(self):
    'Checks whether execution of composed SQL string is traced'
    db = self._get_conn()
    with db.cursor() as cur:
        cur.execute('CREATE TEMP TABLE test (id serial PRIMARY KEY, name varchar(12) NOT NULL UNIQUE);')
        cur.execute('INSERT INTO test (name) VALUES (%s);', ('test_case',))
        spans = self.get_spans()
        assert (len(spans) == 2)
        self.reset()
        query = SQL('select {}, {} from {}').format(Identifier('id'), Identifier('name'), Identifier('test'))
        cur.execute(query=query)
        rows = cur.fetchall()
        assert (rows == [(1, 'test_case')])
        assert_is_measured(self.get_root_span())
        self.assert_structure(dict(name='postgres.query', resource=query.as_string(db)))
Checks whether execution of composed SQL string is traced
@skipIf((PSYCOPG2_VERSION < (2, 7)), 'SQL string composition not available in psycopg2<2.7')
def test_composed_query_identifier(self):
    
    db = self._get_conn()
    with db.cursor() as cur:
        cur.execute('CREATE TEMP TABLE test (id serial PRIMARY KEY, name varchar(12) NOT NULL UNIQUE);')
        cur.execute('INSERT INTO test (name) VALUES (%s);', ('test_case',))
        spans = self.get_spans()
        assert (len(spans) == 2)
        self.reset()
        query = SQL('select {}, {} from {}').format(Identifier('id'), Identifier('name'), Identifier('test'))
        cur.execute(query=query)
        rows = cur.fetchall()
        assert (rows == [(1, 'test_case')])
        assert_is_measured(self.get_root_span())
        self.assert_structure(dict(name='postgres.query', resource=query.as_string(db)))